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
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
|
package gitrepo
import (
"context"
"crypto/sha256"
"errors"
"fmt"
"os"
"path"
"sort"
"strings"
"sync"
"github.com/go-git/go-billy/v5"
"github.com/go-git/go-billy/v5/osfs"
"github.com/go-git/go-billy/v5/util"
"github.com/go-git/go-git/v5/plumbing/protocol/packp"
"github.com/go-git/go-git/v5/plumbing/storer"
"github.com/go-git/go-git/v5/storage/filesystem"
"gitweb/internal/githttp"
"gitweb/internal/store"
)
// ErrConcurrentModification is returned when a repo's store Sequence changed
// between materialization and write-back, meaning the tree was modified out of
// band during the push. The push is abandoned without writing anything.
var ErrConcurrentModification = errors.New("remote repository changed during push; please retry")
// defaultPushConcurrency caps pushes in flight across all repos; each one
// holds a full private copy of its repo on disk while it runs.
const defaultPushConcurrency = 2
// Pusher applies a git receive-pack (push) to an upspin-stored repo. It
// materializes a private, writable copy of the repo, runs the push against it
// with go-git, and writes the changed files back to the store. Pushes to the
// same repo are serialized, and total pushes in flight are bounded.
type Pusher struct {
st store.WriteStore
cache *Cache
sem chan struct{} // bounds concurrent pushes
mu sync.Mutex
locks map[string]*sync.Mutex
}
// NewPusher returns a Pusher writing through st and invalidating c on success.
func NewPusher(st store.WriteStore, c *Cache) *Pusher {
return &Pusher{
st: st,
cache: c,
sem: make(chan struct{}, defaultPushConcurrency),
locks: map[string]*sync.Mutex{},
}
}
func (p *Pusher) lockFor(repoPath string) *sync.Mutex {
p.mu.Lock()
defer p.mu.Unlock()
l := p.locks[repoPath]
if l == nil {
l = &sync.Mutex{}
p.locks[repoPath] = l
}
return l
}
// noPackfileWriter hides the storer's optional PackfileWriter capability from
// the receive-pack session. With it visible, go-git copies the client's
// packfile verbatim into objects/pack, and that path cannot resolve the
// REF_DELTA entries of a thin pack — which git sends on every incremental
// push, deltifying against objects that live in the repo, not in the pack —
// so such pushes fail with "reference delta not found". Hidden, go-git parses
// the pack against the storer, resolving thin deltas and storing the new
// objects loose.
type noPackfileWriter struct{ storer.Storer }
// ReceivePack runs the push described by req against repoPath and writes the
// result back to the store. The returned ReportStatus should be sent to the
// client even when err is non-nil (it carries per-ref status).
func (p *Pusher) ReceivePack(ctx context.Context, repoPath string, req *packp.ReferenceUpdateRequest) (*packp.ReportStatus, error) {
lock := p.lockFor(repoPath)
lock.Lock()
defer lock.Unlock()
// Bound pushes in flight (each holds a private on-disk repo copy). Taken
// after the per-repo lock: holders always release, so no cycle.
select {
case p.sem <- struct{}{}:
defer func() { <-p.sem }()
case <-ctx.Done():
return nil, ctx.Err()
}
seq0, err := p.st.Seq(repoPath)
if err != nil {
return nil, err
}
// A private on-disk copy: never the shared read cache, which serves
// concurrent readers and must not be mutated.
dir, err := os.MkdirTemp(p.cache.dir, "push-")
if err != nil {
return nil, err
}
defer removeDir(dir)
fs := osfs.New(dir)
before, err := materialize(p.st, repoPath, fs, p.cache.maxRepo)
if err != nil {
return nil, err
}
sto := filesystem.NewStorage(fs, p.cache.objects)
sess, err := githttp.NewReceivePackSession(noPackfileWriter{sto})
if err != nil {
return nil, err
}
defer sess.Close()
rs, rpErr := sess.ReceivePack(ctx, req)
if rpErr != nil {
// Unpack or ref update failed; the storer may be in a partial state, so
// do not write anything back. The report status still goes to the client.
return rs, rpErr
}
after, err := walkManifest(fs, p.cache.maxRepo)
if err != nil {
return failReport(rs, err), err
}
// Guard against an out-of-band change to the tree since we materialized.
// The per-repo lock already excludes other pushes through this server.
seq1, err := p.st.Seq(repoPath)
if err != nil {
return failReport(rs, err), err
}
if seq1 != seq0 {
return failReport(rs, ErrConcurrentModification), ErrConcurrentModification
}
if err := p.writeBack(fs, before, after, repoPath); err != nil {
return failReport(rs, err), err
}
p.cache.Invalidate(repoPath)
return rs, nil
}
// failReport downgrades every "ok" per-ref status in rs to err's message. The
// session updated only the private copy; when persisting to the store fails,
// the client must see its refs rejected, not an "ok" report for updates that
// were never written. The message is flattened to one line so it survives
// pkt-line framing.
func failReport(rs *packp.ReportStatus, err error) *packp.ReportStatus {
if rs == nil {
return nil
}
msg := strings.Join(strings.Fields(err.Error()), " ")
for _, cs := range rs.CommandStatuses {
if cs.Error() == nil { // "ok"
cs.Status = msg
}
}
return rs
}
// writeBack applies the difference between the before and after manifests to the
// store: new/changed files are written objects-first then refs, and files that
// disappeared are deleted last. This ordering keeps a concurrent reader's view
// consistent — it may briefly see extra unreachable objects, never a ref that
// points at a missing object.
func (p *Pusher) writeBack(fs billy.Filesystem, before, after Manifest, repoPath string) error {
var objectPuts, refPuts, dels []string
for name, info := range after {
if old, ok := before[name]; ok && old.Sum == info.Sum {
continue // unchanged
}
if strings.HasPrefix(name, "objects/") {
objectPuts = append(objectPuts, name)
} else {
refPuts = append(refPuts, name)
}
}
for name := range before {
if _, ok := after[name]; !ok {
dels = append(dels, name)
}
}
// Deterministic order (aids testing and reproducibility).
sort.Strings(objectPuts)
sort.Strings(refPuts)
sort.Strings(dels)
put := func(name string) error {
data, err := util.ReadFile(fs, name)
if err != nil {
return err
}
return p.st.Put(path.Join(repoPath, name), data)
}
for _, name := range objectPuts {
if err := put(name); err != nil {
return fmt.Errorf("push write object %s: %w", name, err)
}
}
for _, name := range refPuts {
if err := put(name); err != nil {
return fmt.Errorf("push write ref %s: %w", name, err)
}
}
for _, name := range dels {
if err := p.st.Delete(path.Join(repoPath, name)); err != nil && err != store.ErrNotFound {
return fmt.Errorf("push delete %s: %w", name, err)
}
}
return nil
}
// walkManifest builds a manifest of every regular file currently in fs,
// enforcing the size cap and skipping go-git's dotgit temp files.
func walkManifest(fs billy.Filesystem, maxBytes int64) (Manifest, error) {
man := Manifest{}
var total int64
err := util.Walk(fs, "/", func(p string, info os.FileInfo, err error) error {
if err != nil {
return err
}
if info.IsDir() {
return nil
}
rel := strings.TrimPrefix(p, "/")
base := path.Base(rel)
if strings.HasPrefix(base, "tmp_") || strings.HasPrefix(base, "._packed-refs") {
return nil // go-git dotgit scratch file, not part of the repo
}
total += info.Size()
if total > maxBytes {
return fmt.Errorf("%w (limit %d bytes)", ErrRepoTooLarge, maxBytes)
}
data, err := util.ReadFile(fs, rel)
if err != nil {
return err
}
man[rel] = FileInfo{Size: info.Size(), Sum: sha256.Sum256(data)}
return nil
})
if err != nil {
return nil, err
}
return man, nil
}
|