1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package gitcli
4
5import (
6 "bufio"
7 "bytes"
8 "context"
9 "errors"
10 "fmt"
11 "io"
12 "os/exec"
13 "path/filepath"
14 "slices"
15 "strconv"
16 "strings"
17 "sync"
18 "time"
19
20 "github.com/bnema/zerowrap"
21)
22
23// ErrClosed is returned by reads after Close.
24var ErrClosed = errors.New("gitcli: reader is closed")
25
26// errObjectMissing is what a cat-file lookup answers for an unknown name.
27var errObjectMissing = errors.New("object missing")
28
29// procError is an I/O or protocol failure of a cat-file process: the process
30// is dropped and the request may be retried on a fresh one.
31type procError struct{ err error }
32
33func (e *procError) Error() string { return "git cat-file: " + e.err.Error() }
34func (e *procError) Unwrap() error { return e.err }
35
36// Defaults of the process pool.
37const (
38 defaultIdleTimeout = 60 * time.Second
39 defaultMaxProcesses = 16
40 // maxProcsPerRepo bounds the processes one repository may hold. A
41 // popular repository spawns more than one only under contention.
42 maxProcsPerRepo = 4
43)
44
45// catPool keeps long-lived `git cat-file --batch-command` processes, a few per
46// repository.
47//
48// - A process serves one request at a time. A request takes an idle process
49// of the repository; if all are busy and the repository holds fewer than
50// perRepo, a new one is started (so a repository nobody reads
51// concurrently keeps a single process); otherwise the request waits, in
52// order, for a process to be released.
53// - A process nobody used for the idle timeout is killed.
54// - At most max processes are kept, all repositories together. Starting
55// another one first evicts the least recently used idle process; a
56// process that is in use is never evicted. When none can be evicted, a
57// repository that has a process waits for it, and one that has none starts
58// one anyway, so the bound is soft only when more than max repositories
59// are read at the same moment.
60// - A process that fails, or whose request was cancelled (its output is then
61// in an unknown state), is killed and dropped; the next request starts a
62// fresh one, and a request that hit a dead process is retried once.
63type catPool struct {
64 bin string
65 env []string
66 idle time.Duration
67 max int // processes, all repositories together
68 perRepo int // processes of one repository
69
70 mu sync.Mutex
71 procs map[string][]*catProc // by repository directory
72 total int // live processes in procs
73 waiters map[string][]chan *catProc
74 streams map[*blobStream]struct{} // one-shot blob processes, see OpenBlob
75 closed bool
76 wg sync.WaitGroup // reapers of started processes
77}
78
79// catProc is one cat-file process. Fields below the line are guarded by
80// catPool.mu.
81type catProc struct {
82 dir string
83 log zerowrap.Logger // of the request that started the process
84 cmd *exec.Cmd
85 in io.WriteCloser
86 out *bufio.Reader
87 stderr *tailBuffer
88
89 broken bool // set by the holder on a protocol error
90
91 busy bool // a request holds the process
92 lastUsed time.Time
93 timer *time.Timer
94 gen uint64
95 dead bool
96}
97
98func newCatPool(bin string, env []string, idle time.Duration, maxProcs int) *catPool {
99 if idle <= 0 {
100 idle = defaultIdleTimeout
101 }
102 if maxProcs <= 0 {
103 maxProcs = defaultMaxProcesses
104 }
105 return &catPool{
106 bin: bin, env: env, idle: idle, max: maxProcs, perRepo: min(maxProcsPerRepo, maxProcs),
107 procs: make(map[string][]*catProc),
108 waiters: make(map[string][]chan *catProc),
109 streams: make(map[*blobStream]struct{}),
110 }
111}
112
113// do runs fn with exclusive use of a cat-file process of the repository in
114// dir, starting it if needed. fn must not keep the connection after returning.
115func (p *catPool) do(ctx context.Context, dir string, fn func(*catConn) error) error {
116 for attempt := 0; ; attempt++ {
117 cp, err := p.acquire(ctx, dir)
118 if err != nil {
119 return err
120 }
121 err = p.run(ctx, cp, fn)
122 var pe *procError
123 if !errors.As(err, &pe) {
124 return err
125 }
126 switch {
127 case p.isClosed():
128 return ErrClosed
129 case attempt == 0 && ctx.Err() == nil:
130 continue // the process died while idle: retry once on a fresh one
131 }
132 return err
133 }
134}
135
136func (p *catPool) run(ctx context.Context, cp *catProc, fn func(*catConn) error) error {
137 // A cancelled request leaves the process mid-answer: kill it.
138 stop := context.AfterFunc(ctx, func() {
139 p.mu.Lock()
140 p.dropLocked(cp)
141 p.mu.Unlock()
142 })
143 err := fn(&catConn{cp: cp})
144 cancelled := !stop()
145 if cancelled && err != nil {
146 err = ctx.Err()
147 }
148 p.release(cp, cp.broken || cancelled)
149 return err
150}
151
152func (p *catPool) isClosed() bool {
153 p.mu.Lock()
154 defer p.mu.Unlock()
155 return p.closed
156}
157
158// acquire returns a process of dir that no other request uses.
159func (p *catPool) acquire(ctx context.Context, dir string) (*catProc, error) {
160 for {
161 if err := ctx.Err(); err != nil {
162 return nil, err
163 }
164 p.mu.Lock()
165 if p.closed {
166 p.mu.Unlock()
167 return nil, ErrClosed
168 }
169 cp, wait, err := p.takeLocked(ctx, dir)
170 if err != nil {
171 p.mu.Unlock()
172 return nil, err
173 }
174 if cp != nil {
175 p.mu.Unlock()
176 return cp, nil
177 }
178 p.waiters[dir] = append(p.waiters[dir], wait)
179 p.mu.Unlock()
180
181 select {
182 case cp := <-wait:
183 if cp == nil {
184 continue // the processes changed: look again
185 }
186 p.mu.Lock()
187 dead := cp.dead
188 p.mu.Unlock()
189 if !dead {
190 return cp, nil
191 }
192 p.release(cp, false) // dropped between the hand-over and now
193 case <-ctx.Done():
194 p.mu.Lock()
195 if !p.unqueueLocked(dir, wait) {
196 // A process was handed over meanwhile: pass it on.
197 p.mu.Unlock()
198 if cp := <-wait; cp != nil {
199 p.release(cp, false)
200 }
201 } else {
202 p.mu.Unlock()
203 }
204 return nil, ctx.Err()
205 }
206 }
207}
208
209// takeLocked marks an idle process of dir busy and returns it, starting one
210// when none is idle and the repository may have more. Otherwise it returns the
211// channel to wait on.
212func (p *catPool) takeLocked(ctx context.Context, dir string) (*catProc, chan *catProc, error) {
213 var best *catProc // the most recently used idle one, so that the others idle out
214 for _, c := range p.procs[dir] {
215 if !c.busy && (best == nil || c.lastUsed.After(best.lastUsed)) {
216 best = c
217 }
218 }
219 if best == nil {
220 own := len(p.procs[dir])
221 if own < p.perRepo && (p.makeRoomLocked() || own == 0) {
222 var err error
223 if best, err = p.startLocked(ctx, dir); err != nil {
224 return nil, nil, err
225 }
226 }
227 }
228 if best == nil {
229 return nil, make(chan *catProc, 1), nil
230 }
231 best.busy = true
232 best.gen++
233 if best.timer != nil {
234 best.timer.Stop()
235 best.timer = nil
236 }
237 return best, nil, nil
238}
239
240// makeRoomLocked evicts least recently used idle processes until another one
241// fits, and reports whether it does.
242func (p *catPool) makeRoomLocked() bool {
243 for p.total >= p.max {
244 var victim *catProc
245 for _, procs := range p.procs {
246 for _, c := range procs {
247 if !c.busy && (victim == nil || c.lastUsed.Before(victim.lastUsed)) {
248 victim = c
249 }
250 }
251 }
252 if victim == nil {
253 return false
254 }
255 victim.log.Debug().Str("dir", victim.dir).Msg("cat-file pool full, evicting the least recently used")
256 p.dropLocked(victim)
257 }
258 return true
259}
260
261// unqueueLocked removes a waiter and reports whether it was still queued.
262func (p *catPool) unqueueLocked(dir string, wait chan *catProc) bool {
263 q := p.waiters[dir]
264 for i, w := range q {
265 if w == wait {
266 q = append(q[:i:i], q[i+1:]...)
267 if len(q) == 0 {
268 delete(p.waiters, dir)
269 } else {
270 p.waiters[dir] = q
271 }
272 return true
273 }
274 }
275 return false
276}
277
278// release ends one use of cp. A broken process is dropped. A working one goes
279// to the request that waited longest, or else starts its idle countdown.
280func (p *catPool) release(cp *catProc, broken bool) {
281 p.mu.Lock()
282 defer p.mu.Unlock()
283 if broken || cp.dead {
284 cp.busy = false
285 p.dropLocked(cp)
286 return
287 }
288 if q := p.waiters[cp.dir]; len(q) > 0 && !p.closed {
289 w := q[0]
290 if len(q) == 1 {
291 delete(p.waiters, cp.dir)
292 } else {
293 p.waiters[cp.dir] = q[1:]
294 }
295 cp.gen++
296 w <- cp // stays busy: handed over
297 return
298 }
299 cp.busy = false
300 if !p.closed {
301 cp.lastUsed = time.Now()
302 gen := cp.gen
303 cp.timer = time.AfterFunc(p.idle, func() { p.expire(cp, gen) })
304 }
305}
306
307// expire kills cp if it has not been used since the timer was armed.
308func (p *catPool) expire(cp *catProc, gen uint64) {
309 p.mu.Lock()
310 defer p.mu.Unlock()
311 if cp.gen != gen || cp.busy || cp.dead {
312 return
313 }
314 cp.log.Debug().Str("dir", cp.dir).Msg("cat-file idle, stopping")
315 p.dropLocked(cp)
316}
317
318// dropLocked removes cp from the pool and kills it. The reaper goroutine
319// collects the process. Requests waiting for the repository look again.
320func (p *catPool) dropLocked(cp *catProc) {
321 if cp.dead {
322 return
323 }
324 cp.dead = true
325 procs := p.procs[cp.dir]
326 for i, c := range procs {
327 if c == cp {
328 procs = append(procs[:i:i], procs[i+1:]...)
329 p.total--
330 break
331 }
332 }
333 if len(procs) == 0 {
334 delete(p.procs, cp.dir)
335 } else {
336 p.procs[cp.dir] = procs
337 }
338 if cp.timer != nil {
339 cp.timer.Stop()
340 cp.timer = nil
341 }
342 _ = cp.in.Close()
343 _ = cp.cmd.Process.Kill()
344 for _, w := range p.waiters[cp.dir] {
345 w <- nil
346 }
347 delete(p.waiters, cp.dir)
348}
349
350func (p *catPool) startLocked(ctx context.Context, dir string) (*catProc, error) {
351 cmd := exec.Command(p.bin, "--git-dir="+dir, "--no-replace-objects", "--no-optional-locks", "cat-file", "--batch-command="+catFormat, "-Z")
352 cmd.Env = p.env
353 cmd.Dir = filepath.Dir(dir)
354 stderr := &tailBuffer{max: 4096}
355 cmd.Stderr = stderr
356 in, err := cmd.StdinPipe()
357 if err != nil {
358 return nil, fmt.Errorf("start git cat-file: %w", err)
359 }
360 stdout, err := cmd.StdoutPipe()
361 if err != nil {
362 return nil, fmt.Errorf("start git cat-file: %w", err)
363 }
364 if err := cmd.Start(); err != nil {
365 return nil, fmt.Errorf("start git cat-file: %w", err)
366 }
367 cp := &catProc{
368 dir: dir, log: zerowrap.FromCtx(ctx), cmd: cmd, in: in, out: bufio.NewReaderSize(stdout, 64*1024),
369 stderr: stderr, lastUsed: time.Now(),
370 }
371 p.procs[dir] = append(p.procs[dir], cp)
372 p.total++
373 p.wg.Add(1)
374 go func() {
375 defer p.wg.Done()
376 _ = cmd.Wait()
377 }()
378 cp.log.Debug().Str("dir", dir).Int("pid", cmd.Process.Pid).Msg("cat-file started")
379 return cp, nil
380}
381
382// forget kills the processes of the repository in dir, if any.
383func (p *catPool) forget(dir string) {
384 p.mu.Lock()
385 defer p.mu.Unlock()
386 for _, cp := range slices.Clone(p.procs[dir]) {
387 p.dropLocked(cp)
388 }
389}
390
391// close kills every process and waits, until ctx ends, for them to be
392// collected. Later requests fail with ErrClosed.
393func (p *catPool) close(ctx context.Context) error {
394 p.mu.Lock()
395 p.closed = true
396 for _, procs := range p.procs {
397 for _, cp := range slices.Clone(procs) {
398 p.dropLocked(cp)
399 }
400 }
401 streams := make([]*blobStream, 0, len(p.streams))
402 for s := range p.streams {
403 streams = append(streams, s)
404 }
405 p.mu.Unlock()
406 for _, s := range streams {
407 s.kill()
408 }
409
410 done := make(chan struct{})
411 go func() { p.wg.Wait(); close(done) }()
412 select {
413 case <-done:
414 return nil
415 case <-ctx.Done():
416 return fmt.Errorf("wait for git cat-file processes: %w", ctx.Err())
417 }
418}
419
420// tailBuffer keeps the first max bytes written to it, for error messages.
421type tailBuffer struct {
422 mu sync.Mutex
423 max int
424 buf bytes.Buffer
425}
426
427func (b *tailBuffer) Write(p []byte) (int, error) {
428 b.mu.Lock()
429 defer b.mu.Unlock()
430 if room := b.max - b.buf.Len(); room > 0 {
431 b.buf.Write(p[:min(room, len(p))])
432 }
433 return len(p), nil
434}
435
436func (b *tailBuffer) reset() {
437 b.mu.Lock()
438 b.buf.Reset()
439 b.mu.Unlock()
440}
441
442func (b *tailBuffer) String() string {
443 b.mu.Lock()
444 defer b.mu.Unlock()
445 s := strings.TrimSpace(b.buf.String())
446 if len(s) > 512 {
447 s = s[:512] + "…"
448 }
449 return s
450}
451
452// catInfo is the header git prints for an object.
453type catInfo struct {
454 ID string
455 Type string
456 Size int64
457 // Mode is the Git mode of a "<tree-ish>:<path>" object, 0 otherwise.
458 Mode uint32
459}
460
461// catFormat is the answer format of `info` and `contents`.
462const catFormat = "%(objectname) %(objecttype) %(objectsize) %(objectmode)"
463
464// catConn is exclusive use of one cat-file process. Names are objects
465// expressions such as "<commit>:<path>" or "<ref>^{commit}"; they must not
466// contain NUL or line breaks.
467type catConn struct{ cp *catProc }
468
469func (c *catConn) fail(err error) error {
470 c.cp.broken = true
471 if s := c.cp.stderr.String(); s != "" {
472 err = fmt.Errorf("%w: %s", err, s)
473 }
474 return &procError{err: err}
475}
476
477func (c *catConn) send(cmd, name string) error {
478 if strings.ContainsAny(name, "\x00\n") {
479 return fmt.Errorf("invalid object name %q", name)
480 }
481 c.cp.stderr.reset()
482 if _, err := io.WriteString(c.cp.in, cmd+" "+name+"\x00"); err != nil {
483 return c.fail(err)
484 }
485 return nil
486}
487
488// header reads the answer line of a request: the object header, or
489// errObjectMissing.
490func (c *catConn) header(name string) (catInfo, error) {
491 line, err := c.cp.out.ReadString(0)
492 if err != nil {
493 return catInfo{}, c.fail(err)
494 }
495 line = strings.TrimSuffix(line, "\x00")
496 if strings.HasSuffix(line, " missing") || strings.HasSuffix(line, " ambiguous") {
497 return catInfo{}, fmt.Errorf("%w: %s", errObjectMissing, name)
498 }
499 f := strings.Split(line, " ")
500 if len(f) != 4 || len(f[0]) != 40 {
501 return catInfo{}, c.fail(fmt.Errorf("unexpected answer %q", truncateText(line)))
502 }
503 size, err := strconv.ParseInt(f[2], 10, 64)
504 if err != nil || size < 0 {
505 return catInfo{}, c.fail(fmt.Errorf("unexpected size in %q", truncateText(line)))
506 }
507 var mode uint64
508 if f[3] != "" {
509 if mode, err = strconv.ParseUint(f[3], 8, 32); err != nil {
510 return catInfo{}, c.fail(fmt.Errorf("unexpected mode in %q", truncateText(line)))
511 }
512 }
513 return catInfo{ID: f[0], Type: f[1], Size: size, Mode: uint32(mode)}, nil
514}
515
516// info returns the header of an object without its content.
517func (c *catConn) info(name string) (catInfo, error) {
518 if err := c.send("info", name); err != nil {
519 return catInfo{}, err
520 }
521 return c.header(name)
522}
523
524// contents returns an object with its content. It fails if the object is
525// larger than max, after which the process is dropped.
526func (c *catConn) contents(name string, max int64) (catInfo, []byte, error) {
527 if err := c.send("contents", name); err != nil {
528 return catInfo{}, nil, err
529 }
530 h, err := c.header(name)
531 if err != nil {
532 return catInfo{}, nil, err
533 }
534 if h.Size > max {
535 c.cp.broken = true
536 return catInfo{}, nil, fmt.Errorf("object %s is %d bytes, more than %d", name, h.Size, max)
537 }
538 data := make([]byte, h.Size)
539 if _, err := io.ReadFull(c.cp.out, data); err != nil {
540 return catInfo{}, nil, c.fail(err)
541 }
542 if b, err := c.cp.out.ReadByte(); err != nil || b != 0 {
543 return catInfo{}, nil, c.fail(fmt.Errorf("missing terminator after %s", name))
544 }
545 return h, data, nil
546}
547
548func truncateText(s string) string {
549 if len(s) > 80 {
550 return s[:80] + "…"
551 }
552 return s
553}
554
555// addStream registers a one-shot blob process so that close kills it.
556func (p *catPool) addStream(s *blobStream) error {
557 p.mu.Lock()
558 defer p.mu.Unlock()
559 if p.closed {
560 return ErrClosed
561 }
562 p.streams[s] = struct{}{}
563 return nil
564}
565
566func (p *catPool) removeStream(s *blobStream) {
567 p.mu.Lock()
568 delete(p.streams, s)
569 p.mu.Unlock()
570}