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