1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
|
package gitpipe
import (
"bufio"
"bytes"
"context"
"errors"
"fmt"
"io"
"gitlab.com/gitlab-org/gitaly/v14/internal/git"
"gitlab.com/gitlab-org/gitaly/v14/internal/git/catfile"
"gitlab.com/gitlab-org/gitaly/v14/internal/git/localrepo"
)
// CatfileInfoResult is a result for the CatfileInfo pipeline step.
type CatfileInfoResult struct {
// err is an error which occurred during execution of the pipeline.
err error
// ObjectName is the object name as received from the revlistResultChan.
ObjectName []byte
// ObjectInfo is the object info of the object.
ObjectInfo *catfile.ObjectInfo
}
type catfileInfoConfig struct {
skipResult func(*catfile.ObjectInfo) bool
}
// CatfileInfoOption is an option for the CatfileInfo and CatfileInfoAllObjects pipeline steps.
type CatfileInfoOption func(cfg *catfileInfoConfig)
// WithSkipCatfileInfoResult will execute the given function for each ObjectInfo processed by the
// pipeline. If the callback returns `true`, then the object will be skipped and not passed down the
// pipeline.
func WithSkipCatfileInfoResult(skipResult func(*catfile.ObjectInfo) bool) CatfileInfoOption {
return func(cfg *catfileInfoConfig) {
cfg.skipResult = skipResult
}
}
// CatfileInfo processes revlistResults from the given channel and extracts object information via
// `git cat-file --batch-check`. The returned channel will contain all processed catfile info
// results. Any error received via the channel or encountered in this step will cause the pipeline
// to fail. Context cancellation will gracefully halt the pipeline.
func CatfileInfo(
ctx context.Context,
objectInfoReader catfile.ObjectInfoReader,
it ObjectIterator,
opts ...CatfileInfoOption,
) CatfileInfoIterator {
var cfg catfileInfoConfig
for _, opt := range opts {
opt(&cfg)
}
resultChan := make(chan CatfileInfoResult)
go func() {
defer close(resultChan)
for it.Next() {
objectInfo, err := objectInfoReader.Info(ctx, it.ObjectID().Revision())
if err != nil {
sendCatfileInfoResult(ctx, resultChan, CatfileInfoResult{
err: fmt.Errorf("retrieving object info for %q: %w", it.ObjectID(), err),
})
return
}
if cfg.skipResult != nil && cfg.skipResult(objectInfo) {
continue
}
if isDone := sendCatfileInfoResult(ctx, resultChan, CatfileInfoResult{
ObjectName: it.ObjectName(),
ObjectInfo: objectInfo,
}); isDone {
return
}
}
if err := it.Err(); err != nil {
sendCatfileInfoResult(ctx, resultChan, CatfileInfoResult{err: err})
return
}
}()
return &catfileInfoIterator{
ch: resultChan,
}
}
// CatfileInfoAllObjects enumerates all Git objects part of the repository's object directory and
// extracts their object info via `git cat-file --batch-check`. The returned channel will contain
// all processed results. Any error encountered during execution of this pipeline step will cause
// the pipeline to fail. Context cancellation will gracefully halt the pipeline. Note that with this
// pipeline step, the resulting catfileInfoResults will never have an object name.
func CatfileInfoAllObjects(
ctx context.Context,
repo *localrepo.Repo,
opts ...CatfileInfoOption,
) CatfileInfoIterator {
var cfg catfileInfoConfig
for _, opt := range opts {
opt(&cfg)
}
resultChan := make(chan CatfileInfoResult)
go func() {
defer close(resultChan)
var stderr bytes.Buffer
cmd, err := repo.Exec(ctx, git.SubCmd{
Name: "cat-file",
Flags: []git.Option{
git.Flag{Name: "--batch-all-objects"},
git.Flag{Name: "--batch-check"},
git.Flag{Name: "--buffer"},
git.Flag{Name: "--unordered"},
},
}, git.WithStderr(&stderr))
if err != nil {
sendCatfileInfoResult(ctx, resultChan, CatfileInfoResult{
err: fmt.Errorf("spawning cat-file failed: %w", err),
})
return
}
reader := bufio.NewReader(cmd)
for {
objectInfo, err := catfile.ParseObjectInfo(reader)
if err != nil {
if errors.Is(err, io.EOF) {
break
}
sendCatfileInfoResult(ctx, resultChan, CatfileInfoResult{
err: fmt.Errorf("parsing object info: %w", err),
})
return
}
if cfg.skipResult != nil && cfg.skipResult(objectInfo) {
continue
}
if isDone := sendCatfileInfoResult(ctx, resultChan, CatfileInfoResult{
ObjectInfo: objectInfo,
}); isDone {
return
}
}
if err := cmd.Wait(); err != nil {
sendCatfileInfoResult(ctx, resultChan, CatfileInfoResult{
err: fmt.Errorf("cat-file failed: %w, stderr: %q", err, stderr),
})
return
}
}()
return &catfileInfoIterator{
ch: resultChan,
}
}
func sendCatfileInfoResult(ctx context.Context, ch chan<- CatfileInfoResult, result CatfileInfoResult) bool {
// In case the context has been cancelled, we have a race between observing an error from
// the killed Git process and observing the context cancellation itself. But if we end up
// here because of cancellation of the Git process, we don't want to pass that one down the
// pipeline but instead just stop the pipeline gracefully. We thus have this check here up
// front to error messages from the Git process.
select {
case <-ctx.Done():
return true
default:
}
select {
case ch <- result:
return false
case <-ctx.Done():
return true
}
}
|