1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package ermitegit
4
5import (
6 "container/list"
7 "context"
8 "errors"
9 "fmt"
10 "io/fs"
11 "os"
12 "sync"
13 "sync/atomic"
14 "syscall"
15 "time"
16
17 git "bnema.dev/ermite-git"
18
19 "bnema.dev/ermite/internal/domain"
20)
21
22// DefaultMaxRepos is the number of repositories kept open when Config.MaxRepos
23// is zero.
24const DefaultMaxRepos = 64
25
26// DefaultRefreshEvery is how long a repository goes without a Refresh when
27// Config.RefreshEvery is zero. See Config.RefreshEvery for what depends on it.
28// A write that goes through ermite refreshes the repository at once (see
29// Adapter.Refresh), so this is only the safety net for the other writers.
30const DefaultRefreshEvery = 5 * time.Second
31
32// DefaultIdleTTL is how long a repository may go unused before it is closed,
33// when Config.IdleTTL is zero.
34const DefaultIdleTTL = 10 * time.Minute
35
36// DefaultDeltaCacheSize is the budget of the cache of delta bases of each open
37// repository when Config.DeltaCacheSize is zero (the library's own is 32 MiB).
38const DefaultDeltaCacheSize = 8 << 20
39
40// DefaultMaxObjectSize is the largest object a repository holds in memory when
41// Config.MaxObjectSize is zero (the library's own is 128 MiB). A bigger one
42// is read by the wrapped reader.
43const DefaultMaxObjectSize = 16 << 20
44
45// unsupportedRetry is how long a repository that ermite-git cannot read is
46// served by git before Open is tried again (a repack, or the removal of a
47// replace ref, can make it readable).
48const unsupportedRetry = time.Minute
49
50// entry is the state of one repository directory in the cache: an open
51// *git.Repo, or the reason it is not readable by ermite-git.
52//
53// The fields below the line belong to the cache and are guarded by its mutex.
54type entry struct {
55 dir string
56 ino uint64 // of the directory: a new repository under the same name differs
57 dev uint64
58 once sync.Once
59 // opened is the result of git.Open, published by the Once; nil before.
60 opened atomic.Pointer[opened]
61
62 refreshed atomic.Int64 // unix nanoseconds of the last Refresh
63
64 elem *list.Element
65 users int
66 lastUsed time.Time // of the last get or release: the LRU is ordered by it
67 dropped bool // out of the cache: close when the last user is done
68}
69
70// opened is what git.Open answered for an entry.
71type opened struct {
72 repo *git.Repo // nil when unsupported
73 err error // Open failed with something else than ErrUnsupported
74 at time.Time
75}
76
77// handle is a use of an entry; release it when the read is over. repo is nil
78// when git must serve the read.
79type handle struct {
80 c *cache
81 e *entry
82 repo *git.Repo
83 peek bool // from peek: its release is not a use
84}
85
86func (h *handle) release() { h.c.release(h.e, !h.peek) }
87
88// cache is a bounded LRU of open repositories keyed by directory. A repository
89// leaves it by eviction, by idling longer than the TTL, by Forget (the
90// directory was deleted) or because the directory is no longer the same one
91// (checked on each use through its inode); it is closed once the reads that
92// use it are done, by the goroutine that made it leave or that was the last
93// to use it.
94type cache struct {
95 max int
96 refreshEvery time.Duration
97 idleTTL time.Duration // <= 0: never
98 openOpts []git.Option
99 onUnsupport func(ctx context.Context, dir string, name domain.RepoName, cause error)
100
101 mu sync.Mutex
102 entries map[string]*entry
103 lru *list.List // front = most recently used
104 doomed []*entry // dropped without users: closed by unlock
105 closed bool
106
107 stop chan struct{} // closed by close; ends the sweeper
108 swept sync.WaitGroup
109 stopped sync.Once
110}
111
112// cacheConfig is what newCache needs.
113type cacheConfig struct {
114 max int
115 refreshEvery time.Duration
116 idleTTL time.Duration
117 openOpts []git.Option
118 onUnsupport func(context.Context, string, domain.RepoName, error)
119}
120
121func newCache(cfg cacheConfig) *cache {
122 if cfg.max <= 0 {
123 cfg.max = DefaultMaxRepos
124 }
125 c := &cache{
126 max: cfg.max, refreshEvery: cfg.refreshEvery, idleTTL: cfg.idleTTL, openOpts: cfg.openOpts,
127 onUnsupport: cfg.onUnsupport, entries: make(map[string]*entry), lru: list.New(), stop: make(chan struct{}),
128 }
129 if c.idleTTL > 0 {
130 c.swept.Add(1)
131 go c.sweeper()
132 }
133 return c
134}
135
136// sweeper closes the repositories that idle past the TTL. It is the only
137// goroutine of the cache, whatever the number of repositories, and close ends
138// it.
139func (c *cache) sweeper() {
140 defer c.swept.Done()
141 tick := time.NewTicker(max(c.idleTTL/4, time.Millisecond))
142 defer tick.Stop()
143 for {
144 select {
145 case <-c.stop:
146 return
147 case now := <-tick.C:
148 c.sweep(now)
149 }
150 }
151}
152
153// sweep drops the entries nobody used since now-idleTTL. The LRU is ordered by
154// lastUsed, so the walk from the back stops at the first recent entry; one
155// that is in use is not idle and stays.
156func (c *cache) sweep(now time.Time) {
157 c.mu.Lock()
158 for el := c.lru.Back(); el != nil; {
159 prev := el.Prev()
160 e := el.Value.(*entry)
161 if now.Sub(e.lastUsed) <= c.idleTTL {
162 break
163 }
164 if e.users == 0 {
165 c.dropLocked(e)
166 }
167 el = prev
168 }
169 c.unlock()
170}
171
172// unlock releases the mutex, then closes what was dropped without users while
173// it was held: in the calling goroutine, so that when close returns the
174// mappings are released, and not under the mutex, since Close takes the
175// repository's own.
176func (c *cache) unlock() {
177 doomed := c.doomed
178 c.doomed = nil
179 c.mu.Unlock()
180 for _, e := range doomed {
181 e.closeRepo()
182 }
183}
184
185// get returns the entry of dir, opened and refreshed, or a nil repo when
186// ermite-git does not read the repository. A missing directory is
187// domain.ErrNotFound.
188func (c *cache) get(ctx context.Context, name domain.RepoName, dir string) (*handle, error) {
189 fi, err := os.Stat(dir)
190 if err != nil {
191 if errors.Is(err, fs.ErrNotExist) {
192 return nil, domain.ErrNotFound
193 }
194 return nil, fmt.Errorf("repository %q: %w", name, err)
195 }
196 if !fi.IsDir() {
197 return nil, domain.ErrNotFound
198 }
199 dev, ino := identity(fi)
200
201 c.mu.Lock()
202 if c.closed {
203 c.unlock()
204 return nil, ErrClosed
205 }
206 e := c.entries[dir]
207 if e != nil && (e.ino != ino || e.dev != dev || e.stale()) {
208 c.dropLocked(e)
209 e = nil
210 }
211 if e == nil {
212 e = &entry{dir: dir, ino: ino, dev: dev}
213 e.elem = c.lru.PushFront(e)
214 c.entries[dir] = e
215 c.evictLocked()
216 } else {
217 c.lru.MoveToFront(e.elem)
218 }
219 e.users++
220 e.lastUsed = time.Now()
221 c.unlock()
222
223 h := &handle{c: c, e: e}
224 e.once.Do(func() { c.open(ctx, name, e) })
225 op := e.opened.Load()
226 if op.err != nil {
227 c.forgetEntry(e) // the next read tries again
228 h.release()
229 return nil, fmt.Errorf("open repository %q: %w", name, mapOpenErr(op.err))
230 }
231 if op.repo == nil {
232 return h, nil // unsupported
233 }
234 if err := c.refresh(e, op.repo); err != nil {
235 if errors.Is(err, git.ErrUnsupported) {
236 h.fallBack(ctx, name, err)
237 return h, nil
238 }
239 h.release()
240 return nil, fmt.Errorf("refresh repository %q: %w", name, err)
241 }
242 h.repo = op.repo
243 return h, nil
244}
245
246// refresh calls Refresh on the repository of e unless one ran less than
247// refreshEvery ago (a negative interval refreshes on every read). Of the reads
248// that find it due, one wins a compare-and-swap on the time of the last
249// Refresh and runs it; the others go on with the snapshot they have. The
250// library reads loose refs live, stat-checks packed-refs on each ref read and
251// rescans the packs when an object is missing, so a read sees a push whether
252// or not Refresh ran; Refresh is what drops replaced packs, picks up a new
253// commit-graph and notices that the repository became unsupported.
254func (c *cache) refresh(e *entry, repo *git.Repo) error {
255 if c.refreshEvery < 0 {
256 return repo.Refresh()
257 }
258 now := time.Now().UnixNano()
259 last := e.refreshed.Load()
260 if now-last < int64(c.refreshEvery) || !e.refreshed.CompareAndSwap(last, now) {
261 return nil
262 }
263 if err := repo.Refresh(); err != nil {
264 e.refreshed.CompareAndSwap(now, last) // the next read tries again
265 return err
266 }
267 return nil
268}
269
270// stale reports whether a repository that ermite-git did not read is due for
271// another try. The caller holds the cache mutex.
272func (e *entry) stale() bool {
273 op := e.opened.Load()
274 return op != nil && op.repo == nil && op.err == nil && time.Since(op.at) > unsupportedRetry
275}
276
277// open runs git.Open once per entry; concurrent users wait in Once.Do.
278func (c *cache) open(ctx context.Context, name domain.RepoName, e *entry) {
279 op := &opened{at: time.Now()}
280 repo, err := git.Open(e.dir, c.openOpts...)
281 switch {
282 case err == nil:
283 op.repo = repo
284 e.refreshed.Store(op.at.UnixNano()) // just opened: nothing to refresh yet
285 case errors.Is(err, git.ErrUnsupported):
286 c.onUnsupport(ctx, e.dir, name, err)
287 default:
288 op.err = err
289 }
290 e.opened.Store(op)
291}
292
293// mapOpenErr maps the failure of git.Open: a directory that is not a
294// repository is a missing repository.
295func mapOpenErr(err error) error {
296 if errors.Is(err, git.ErrNotFound) {
297 return fmt.Errorf("%w: %w", domain.ErrNotFound, err)
298 }
299 return err
300}
301
302// markUnsupported turns a repository that stopped being readable (a replace
303// ref was pushed, say) into an unsupported one: git serves it from now on.
304func (c *cache) markUnsupported(ctx context.Context, name domain.RepoName, e *entry, cause error) {
305 c.onUnsupport(ctx, e.dir, name, cause)
306 c.mu.Lock()
307 defer c.unlock()
308 if c.entries[e.dir] == e {
309 c.dropLocked(e)
310 n := &entry{dir: e.dir, ino: e.ino, dev: e.dev, lastUsed: time.Now()}
311 n.once.Do(func() { n.opened.Store(&opened{at: time.Now()}) }) // no repo, no error: unsupported
312 n.elem = c.lru.PushFront(n)
313 c.entries[e.dir] = n
314 }
315}
316
317// fallBack records that the repository stopped being readable by ermite-git
318// (a replace ref was pushed, a feature appeared midway): git serves it from
319// now on, and this read too.
320func (h *handle) fallBack(ctx context.Context, name domain.RepoName, cause error) {
321 h.repo = nil
322 h.c.markUnsupported(ctx, name, h.e, cause)
323}
324
325// peek returns a handle on the open repository of dir, or nil when the cache
326// has none (it opens nothing and does not count as a use).
327func (c *cache) peek(dir string) *handle {
328 c.mu.Lock()
329 defer c.unlock()
330 e := c.entries[dir]
331 if c.closed || e == nil {
332 return nil
333 }
334 if op := e.opened.Load(); op == nil || op.repo == nil {
335 return nil
336 }
337 e.users++
338 return &handle{c: c, e: e, repo: e.opened.Load().repo, peek: true}
339}
340
341// refreshNow runs Refresh on the open repository of dir, if the cache has it,
342// whatever the time of the last one.
343func (c *cache) refreshNow(ctx context.Context, name domain.RepoName, dir string) error {
344 h := c.peek(dir)
345 if h == nil {
346 return nil
347 }
348 defer h.release()
349 if err := h.repo.Refresh(); err != nil {
350 if errors.Is(err, git.ErrUnsupported) {
351 h.fallBack(ctx, name, err)
352 return nil
353 }
354 return fmt.Errorf("refresh repository %q: %w", name, err)
355 }
356 h.e.refreshed.Store(time.Now().UnixNano())
357 return nil
358}
359
360func (c *cache) release(e *entry, use bool) {
361 c.mu.Lock()
362 e.users--
363 if e.dropped {
364 if e.users == 0 {
365 c.doomed = append(c.doomed, e)
366 }
367 } else if use {
368 e.lastUsed = time.Now() // a long read is a use, up to its end
369 c.lru.MoveToFront(e.elem)
370 }
371 c.unlock()
372}
373
374// closeRepo closes the repository of an entry that left the cache and has no
375// user. An entry whose open has not finished has no repository yet: its last
376// user closes it.
377func (e *entry) closeRepo() {
378 if op := e.opened.Load(); op != nil && op.repo != nil {
379 _ = op.repo.Close()
380 }
381}
382
383// dropLocked removes e from the cache. The caller holds the mutex and gives it
384// up with unlock, which closes the repository if e has no user; else the last
385// user closes it on release.
386func (c *cache) dropLocked(e *entry) {
387 if e.dropped {
388 return
389 }
390 e.dropped = true
391 c.lru.Remove(e.elem)
392 delete(c.entries, e.dir)
393 if e.users == 0 {
394 c.doomed = append(c.doomed, e)
395 }
396}
397
398// evictLocked drops the least recently used entries beyond the bound.
399func (c *cache) evictLocked() {
400 for c.lru.Len() > c.max {
401 c.dropLocked(c.lru.Back().Value.(*entry))
402 }
403}
404
405// forgetEntry drops e if it is still the entry of its directory.
406func (c *cache) forgetEntry(e *entry) {
407 c.mu.Lock()
408 defer c.unlock()
409 if c.entries[e.dir] == e {
410 c.dropLocked(e)
411 }
412}
413
414// forget drops the entry of dir.
415func (c *cache) forget(dir string) {
416 c.mu.Lock()
417 defer c.unlock()
418 if e := c.entries[dir]; e != nil {
419 c.dropLocked(e)
420 }
421}
422
423// close drops every entry; later gets fail with ErrClosed. The repositories
424// nobody uses are closed when it returns; the others when their last reader
425// is done. It ends the sweeper and waits for it.
426func (c *cache) close() {
427 c.stopped.Do(func() { close(c.stop) })
428 c.swept.Wait()
429 c.mu.Lock()
430 defer c.unlock()
431 c.closed = true
432 for _, e := range c.entries {
433 c.dropLocked(e)
434 }
435}
436
437// size returns the number of cached repositories.
438func (c *cache) size() int {
439 c.mu.Lock()
440 defer c.mu.Unlock()
441 return len(c.entries)
442}
443
444func identity(fi fs.FileInfo) (dev, ino uint64) {
445 if st, ok := fi.Sys().(*syscall.Stat_t); ok {
446 return uint64(st.Dev), st.Ino //nolint:unconvert // Dev is uint32 on some platforms
447 }
448 return 0, 0
449}