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"
13 "os/exec"
14 "sync"
15
16 "bnema.dev/ermite/internal/domain"
17)
18
19// smallBlob is the largest blob read whole, through the long-lived cat-file
20// process. A larger one streams from a process of its own, so that a slow
21// reader never holds the shared process.
22const smallBlob = 256 << 10
23
24// OpenBlob describes the file at path and returns its content as a stream.
25//
26// A blob up to 256 KiB is read through the repository's long-lived cat-file
27// process. A larger one gets a `git cat-file blob` process that lives until
28// the reader is closed, or ctx ends, or the adapter is closed; the memory
29// used is a few buffers, whatever the size. The reader fails with
30// io.ErrUnexpectedEOF if git stops before the announced size.
31//
32//hexcheck:ignore suspicious-business-logic-in-adapter builds domain values from git output; size and binary rules are the port contract
33func (g *Git) OpenBlob(ctx context.Context, repo domain.RepoName, rev domain.Revision, p domain.TreePath) (domain.BlobInfo, io.ReadCloser, error) {
34 dir, err := g.repoDir(repo)
35 if err != nil {
36 return domain.BlobInfo{}, nil, err
37 }
38 if p.IsRoot() {
39 return domain.BlobInfo{}, nil, fmt.Errorf("blob of %q: %w", repo, domain.ErrNotFile)
40 }
41 var (
42 h catInfo
43 data []byte
44 )
45 err = g.cat.do(ctx, dir, func(c *catConn) error {
46 h, data = catInfo{}, nil
47 res, err := resolve(c, rev)
48 if err != nil {
49 return err
50 }
51 h, err = c.info(res.CommitID.String() + ":" + p.String())
52 if notFound(err) {
53 return fmt.Errorf("path %q: %w", p, domain.ErrNotFound)
54 }
55 if err != nil {
56 return err
57 }
58 if h.Type != "blob" || h.Size > smallBlob {
59 return nil
60 }
61 _, data, err = c.contents(h.ID, smallBlob)
62 return err
63 })
64 if err != nil {
65 return domain.BlobInfo{}, nil, fmt.Errorf("blob %q of %q: %w", p, repo, err)
66 }
67 if h.Type != "blob" {
68 return domain.BlobInfo{}, nil, fmt.Errorf("blob %q of %q: %w", p, repo, domain.ErrNotFile)
69 }
70 id, err := domain.ParseObjectID(h.ID)
71 if err != nil {
72 return domain.BlobInfo{}, nil, err
73 }
74 info := domain.BlobInfo{ID: id, Path: p.String(), Size: h.Size, Mode: h.Mode}
75
76 if data != nil || h.Size == 0 {
77 info.SniffContent(data)
78 return info, io.NopCloser(bytes.NewReader(data)), nil
79 }
80
81 s, err := g.startBlobStream(ctx, dir, id, h.Size)
82 if err != nil {
83 return domain.BlobInfo{}, nil, fmt.Errorf("blob %q of %q: %w", p, repo, err)
84 }
85 head, err := s.r.Peek(min(int(h.Size), domain.BinarySniffLen))
86 if err != nil && !errors.Is(err, bufio.ErrBufferFull) {
87 _ = s.Close()
88 return domain.BlobInfo{}, nil, fmt.Errorf("blob %q of %q: %w", p, repo, ctxOr(ctx, err))
89 }
90 info.SniffContent(head)
91 return info, s, nil
92}
93
94func ctxOr(ctx context.Context, err error) error {
95 if cerr := ctx.Err(); cerr != nil {
96 return cerr
97 }
98 return err
99}
100
101// blobStream is the content of a blob read from its own git process.
102type blobStream struct {
103 pool *catPool
104 cmd *exec.Cmd
105 pipe *os.File
106 r *bufio.Reader
107 size int64
108 read int64
109 stop func() bool // detaches the ctx watcher
110
111 done chan struct{} // closed when the process has been collected
112 once sync.Once
113 release func() // gives the stream slot back
114}
115
116// startBlobStream starts the process of a blob stream. The process is counted
117// against Config.MaxBlobStreams from its start until the stream is closed (or
118// the process is collected); there is no queue: a full table is refused with
119// domain.ErrResourceExhausted.
120func (g *Git) startBlobStream(ctx context.Context, dir string, id domain.ObjectID, size int64) (*blobStream, error) {
121 select {
122 case g.blobs <- struct{}{}:
123 default:
124 return nil, fmt.Errorf("too many blob streams open: %w", domain.ErrResourceExhausted)
125 }
126 release := func() { <-g.blobs }
127 pr, pw, err := os.Pipe()
128 if err != nil {
129 release()
130 return nil, fmt.Errorf("start git cat-file: %w", err)
131 }
132 cmd := exec.Command(g.bin, readArgs(dir, "cat-file", "blob", id.String())...)
133 cmd.Env = g.env
134 cmd.Dir = g.dataDir
135 cmd.Stdout = pw
136 if err := cmd.Start(); err != nil {
137 _ = pr.Close()
138 _ = pw.Close()
139 release()
140 return nil, fmt.Errorf("start git cat-file: %w", err)
141 }
142 _ = pw.Close()
143 s := &blobStream{
144 pool: g.cat, cmd: cmd, pipe: pr, r: bufio.NewReaderSize(pr, 64*1024),
145 size: size, done: make(chan struct{}), release: release,
146 }
147 go s.reap()
148 if err := g.cat.addStream(s); err != nil {
149 _ = s.Close()
150 return nil, err
151 }
152 s.stop = context.AfterFunc(ctx, s.kill)
153 return s, nil
154}
155
156// reap collects the process and, with it, the stream slot: a process that is
157// gone no longer counts, whether or not the reader was closed.
158func (s *blobStream) reap() {
159 _ = s.cmd.Wait()
160 s.release()
161 close(s.done)
162}
163
164func (s *blobStream) kill() { _ = s.cmd.Process.Kill() }
165
166// Read implements io.Reader.
167func (s *blobStream) Read(p []byte) (int, error) {
168 n, err := s.r.Read(p)
169 s.read += int64(n)
170 if errors.Is(err, io.EOF) && s.read < s.size {
171 return n, io.ErrUnexpectedEOF
172 }
173 return n, err
174}
175
176// Close stops the process, waits for it and releases the pipe. It is safe to
177// call more than once.
178func (s *blobStream) Close() error {
179 s.once.Do(func() {
180 if s.stop != nil {
181 s.stop()
182 }
183 s.kill()
184 <-s.done
185 _ = s.pipe.Close()
186 s.pool.removeStream(s)
187 })
188 return nil
189}