Git backed by object storage because you can't stop me
git
object-storage
kefka
fix: further optimize pack caching
Signed-off-by: Xe Iaso <me@xeiaso.net>
+1410
-78
+125
docs/plans/git-default-branck-problems.md
+125
docs/plans/git-default-branck-problems.md
···
1
+
# Fix: clone leaves `warning: remote HEAD refers to nonexistent ref, unable to checkout`
2
+
3
+
## Context
4
+
5
+
Cloning a mirror of `golang/go` from objgitd downloads every object and resolves all
6
+
deltas, then aborts the checkout with:
7
+
8
+
```
9
+
warning: remote HEAD refers to nonexistent ref, unable to checkout
10
+
```
11
+
12
+
**Root cause — a `main` vs `master` HEAD mismatch that is never healed.**
13
+
`loadOrInit` (`cmd/objgitd/git_protocol.go:198`) initializes _every_ new repository
14
+
with `git.Init(..., git.WithDefaultBranch(refs/heads/main))`, so its `HEAD` file is
15
+
the symbolic ref `ref: refs/heads/main`. Receive-pack only writes the branch refs the
16
+
client pushed (`updateReferences` in `receivepack.go:287`) and **never touches HEAD**.
17
+
When you push a project whose default branch is not `main` — `golang/go` uses
18
+
`master` — the repo ends up with `refs/heads/master` (plus release branches) but a
19
+
`HEAD` that still points at the nonexistent `refs/heads/main`. HEAD dangles forever.
20
+
21
+
Confirmed empirically (both reproduced this turn):
22
+
23
+
- go-git v6.0.0-alpha.4's advertiser (`plumbing/transport/serve.go:96` `addReferences`)
24
+
resolves HEAD's symbolic target and, on `ErrReferenceNotFound`, **drops HEAD from the
25
+
advertisement entirely** (no `symref=HEAD:...` capability). Pointing HEAD at an
26
+
existing branch instead produces the correct `symref=HEAD:refs/heads/master`.
27
+
- A real bare repo (`git init --bare -b main`, then push only `master`) reproduces the
28
+
exact client warning over `file://`. So the trigger is purely the dangling HEAD;
29
+
it is not a go-git wire bug.
30
+
31
+
This is a generic correctness gap, not specific to `golang/go`: any pushed repo whose
32
+
default branch ≠ `main` is unclonable-to-worktree. Real git hosts (GitHub/GitLab/Gitea)
33
+
repoint HEAD on push; objgitd does not.
34
+
35
+
**Intended outcome:** after a push, and for repos already sitting in the bucket, HEAD
36
+
resolves to a branch that exists, so `git clone` checks out a worktree with no warning.
37
+
38
+
## Approach
39
+
40
+
Add an idempotent **heal HEAD** step and call it (a) on every load and (b) after every
41
+
push. Healing only acts when HEAD is symbolic _and_ its target is missing; a detached
42
+
HEAD, an already-valid HEAD, or a branch-less repo is left untouched. This fixes future
43
+
pushes immediately and lets repos already broken in the bucket (e.g. `golang/go`)
44
+
recover on their next clone — one `HEAD` rewrite, then self-correcting — without a
45
+
re-push.
46
+
47
+
### New helpers (in `cmd/objgitd/git_protocol.go`, next to `loadOrInit`)
48
+
49
+
```go
50
+
// ensureHEAD repoints a repo's HEAD at an existing branch when its symbolic target
51
+
// is missing. objgitd inits every repo with HEAD -> refs/heads/main, but pushing a
52
+
// project whose default branch differs (golang/go uses master) leaves HEAD dangling:
53
+
// clients fetch every object yet cannot check out a worktree. Idempotent and best-
54
+
// effort; a detached/valid HEAD or a branch-less repo is left alone.
55
+
func ensureHEAD(st storage.Storer) error {
56
+
head, err := st.Reference(plumbing.HEAD)
57
+
if err != nil { return err }
58
+
if head.Type() != plumbing.SymbolicReference { return nil } // detached: leave
59
+
if _, err := st.Reference(head.Target()); err == nil { return nil } // already valid
60
+
else if !errors.Is(err, plumbing.ErrReferenceNotFound) { return err }
61
+
target, err := pickDefaultBranch(st)
62
+
if err != nil || target == "" { return err } // no branches: leave
63
+
return st.SetReference(plumbing.NewSymbolicReference(plumbing.HEAD, target))
64
+
}
65
+
66
+
// pickDefaultBranch chooses a branch to point HEAD at: prefer refs/heads/main, then
67
+
// master, then trunk; otherwise the lexicographically smallest branch (deterministic).
68
+
func pickDefaultBranch(st storage.Storer) (plumbing.ReferenceName, error) { ... IterReferences, IsBranch ... }
69
+
```
70
+
71
+
### Wiring (minimal, centralized)
72
+
73
+
1. **New method `(d *daemon) load(repoPath string) (storage.Storer, error)`** — wraps the
74
+
existing `d.loader.Load(&url.URL{Path: repoPath})` and, on success, calls
75
+
`ensureHEAD`, logging (not failing) on heal error so a clone is never broken by a
76
+
transient write failure. Returns the loader's error verbatim (preserves
77
+
`transport.ErrRepositoryNotFound` → 404 semantics).
78
+
79
+
2. **Replace the five direct read-path `d.loader.Load` calls with `d.load`:**
80
+
- `git_protocol.go:147` (git:// upload-pack), `:157` (upload-archive)
81
+
- `ssh.go:192` (ssh upload-pack), `:204` (upload-archive)
82
+
- `http.go:190` (`resolve`, the shared read path for both info/refs advertise and RPC)
83
+
84
+
3. **`loadOrInit` (`git_protocol.go:184`)** — call `d.load` instead of `d.loader.Load`
85
+
for its found path, so the receive-pack advertise phase and pre-push load also heal.
86
+
The create path is unchanged (fresh repo has no branches → `pickDefaultBranch` returns
87
+
"" → HEAD stays `main` until a branch is pushed).
88
+
89
+
4. **Post-push heal in `(d *daemon) receivePack` (`hooks.go:88`)** — after
90
+
`receivePackStreaming` returns `nil`, call `ensureHEAD(st)` (log on error). This makes
91
+
HEAD correct the moment a first push creates branches, so the write lands during a
92
+
push (a write op) rather than during the first subsequent clone.
93
+
94
+
No changes to the vendored go-git fork (`receivepack.go`) are needed.
95
+
96
+
## Critical files
97
+
98
+
- `cmd/objgitd/git_protocol.go` — add `ensureHEAD` + `pickDefaultBranch` + `d.load`; route `loadOrInit` and the two git:// read sites through `d.load`.
99
+
- `cmd/objgitd/ssh.go` — two read sites → `d.load`.
100
+
- `cmd/objgitd/http.go` — `resolve` read site (line 190) → `d.load`.
101
+
- `cmd/objgitd/hooks.go` — `receivePack` calls `ensureHEAD` after a successful receive.
102
+
103
+
## Tests
104
+
105
+
Follow the **xe-go:go-table-driven-tests** skill (per project memory) and the existing
106
+
`tt` table style; reuse `runGit`/`tryGit`/`seedRepo` from `git_protocol_test.go`.
107
+
108
+
1. **Unit (fast, no git binary)** — `ensureHEAD`/`pickDefaultBranch` over a memfs
109
+
`filesystem.Storage`: dangling `HEAD->main` with only `master` present → HEAD repoints
110
+
to `master`; `main` present → unchanged; detached HEAD → unchanged; no branches →
111
+
unchanged; main+master both present → prefers `main`.
112
+
2. **End-to-end (gated on `git` on PATH)** — drive the daemon (HTTP and/or SSH like the
113
+
existing protocol tests): create a repo and push a single `master` branch (no `main`),
114
+
then `git clone` it and assert the worktree checked out (clone exit 0, expected file
115
+
present, no "nonexistent ref" warning) and that `info/refs` advertises
116
+
`symref=HEAD:refs/heads/master`.
117
+
118
+
## Verification
119
+
120
+
- `go build ./...` and `go test ./...` (and `go test ./cmd/objgitd/...` with `git` on PATH).
121
+
- Manual against the live setup (local Garage at `:3903`, bucket `xe-git-repos`,
122
+
`SSH_BIND=:2222`): with `golang/go` already in the bucket, run
123
+
`git clone ssh://localhost:2222/github.com/golang/go.git` and confirm it checks out
124
+
`master` with no warning. The first clone heals HEAD (one `HEAD` write); subsequent
125
+
clones find HEAD already valid.
+301
docs/plans/pack-streaming.md
+301
docs/plans/pack-streaming.md
···
1
+
# Stream pack downloads: serve reads while downloading (don't block clones)
2
+
3
+
## Context
4
+
5
+
Commit `e114543` added `internal/s3fs/packcache.go` to fix clone hangs: immutable
6
+
pack-directory files (`.pack`/`.idx`/`.rev`) are now downloaded **whole** to a
7
+
local temp file on first open and served from disk, collapsing thousands of S3
8
+
`GetObject` round-trips into one. That fixed the hang, but introduced a new
9
+
latency wall: `PackCache.open` blocks inside `e.once.Do` → `download` →
10
+
`io.Copy(tmp, out.Body)` until the **entire** object lands on disk. No byte of
11
+
the pack can be read until the last byte arrives, and every concurrent opener of
12
+
the same key serializes behind that single `sync.Once`. For a multi-hundred-MB
13
+
pack this is dead wait before upload-pack can even start walking the pack header.
14
+
15
+
The goal: **overlap the download with reads.** Stream the S3 body into the temp
16
+
file, advancing a watermark as bytes land, and hand callers a reader
17
+
_immediately_. Reads of already-downloaded byte ranges return at once; reads
18
+
ahead of the watermark block only until that specific range arrives (the S3 body
19
+
is a single sequential stream, so the watermark is monotonic and any requested
20
+
offset below `size` is eventually satisfiable). This turns "wait for the whole
21
+
pack, then serve" into "serve as it streams."
22
+
23
+
Per the design decisions taken up front:
24
+
25
+
- **RAM bound = bounded trailing window ("free read prefix").** We write-through
26
+
to the temp file as we download, so every byte below the watermark is already
27
+
durable on disk. The in-RAM buffer therefore only needs to hold a fixed-size
28
+
_trailing window_ `[n-ringCap, n)`; older bytes are dropped from RAM and served
29
+
from disk on demand. Peak RAM ≈ `inflight_entries × ringCap`, independent of
30
+
pack size. (Reads that fall below the window — e.g. a backward seek before the
31
+
download finishes — re-read from the disk fd, which is correct and cheap via
32
+
the page cache.)
33
+
- **Eviction preserves unlink-while-open.** Eviction `os.Remove`s the temp path
34
+
even while readers (and the in-flight writer) hold it open; they survive via
35
+
their already-open fds on the unlinked inode. To make this work with streaming
36
+
we give the entry **one shared read fd** opened at creation (before any unlink)
37
+
rather than `os.Open`-per-reader, plus a refcount so the fds are closed only
38
+
once the last reader is gone.
39
+
40
+
## Current shape (what changes)
41
+
42
+
All in `internal/s3fs/packcache.go`. The entry/open/download trio is rewritten;
43
+
`isPackCacheable`, `NewPackCache`, `Cleanup`, and the `WithPackCache`/`basic.go`
44
+
wiring stay as-is. `basic.go:77-84` still calls `fs3.packCache.open(...)` and
45
+
gets back a `billy.File`.
46
+
47
+
Today:
48
+
49
+
- `packEntry` = `{key, once, path, size, err, used}` — `once` runs the full
50
+
blocking `download`.
51
+
- `download` = `GetObject` + `io.Copy(tmp, body)` + `tmp.Close()`, fully
52
+
synchronous.
53
+
- `open` blocks in `once.Do`, then `os.Open(e.path)` per reader → `packCachedFile`
54
+
embedding an independent `*os.File`.
55
+
- `evictLocked` unlinks victim temp files; open readers survive via their own fds.
56
+
57
+
## Design
58
+
59
+
### 1. Rework `packEntry` into a streaming entry
60
+
61
+
```go
62
+
type packEntry struct {
63
+
key string
64
+
once sync.Once // guards the header GetObject + pump launch only
65
+
66
+
mu sync.Mutex
67
+
cond *sync.Cond // broadcast as n advances, and when done/err flips
68
+
69
+
wfd *os.File // write side: pump appends sequentially
70
+
rfd *os.File // shared read side: readers ReadAt at offsets < n (survives unlink)
71
+
path string
72
+
73
+
win []byte // trailing RAM window; win covers [winStart, n)
74
+
winStart int64
75
+
n int64 // bytes downloaded+written so far (monotonic watermark)
76
+
size int64 // total, from Content-Length (-1 if unknown)
77
+
done bool // body fully drained, success
78
+
err error // terminal error (header or body)
79
+
80
+
used uint64 // LRU, set on each open
81
+
refs int // live reader handles
82
+
evicted bool // path unlinked; close fds when refs hits 0
83
+
}
84
+
```
85
+
86
+
`ringCap` is a package const (start ~4 MiB; small enough to bound RAM, large
87
+
enough to absorb the read-ahead the scanner does). Optionally surface as a flag
88
+
later — not required for v1.
89
+
90
+
### 2. `open`: launch the pump once, return immediately
91
+
92
+
```go
93
+
func (c *PackCache) open(ctx, client, bucket, key, name) (billy.File, error) {
94
+
// find/create entry under c.mu (unchanged)
95
+
e.once.Do(func() { c.start(ctx, client, bucket, key, e) })
96
+
if e.err != nil { return nil, e.err } // header GetObject failed (e.g. not-found)
97
+
// refs++, seq/used update under c.mu
98
+
return &packCachedFile{e: e, name: name}, nil // NO blocking download
99
+
}
100
+
```
101
+
102
+
`c.start` does the **header** fetch synchronously (so not-found / auth errors
103
+
surface to the caller exactly as today, and `size` is known before the first
104
+
read), then hands the body to a background goroutine:
105
+
106
+
```go
107
+
func (c *PackCache) start(ctx, client, bucket, key, e) {
108
+
out, err := client.GetObject(ctx, &s3.GetObjectInput{Bucket:&bucket, Key:&key})
109
+
observeS3("GetObject", start, err)
110
+
if err != nil { /* notFound→fs.ErrNotExist; drop entry from map; e.err=... */ return }
111
+
tmp, _ := os.CreateTemp(c.dir, "obj-")
112
+
rfd, _ := os.Open(tmp.Name()) // shared read fd, opened before any unlink
113
+
e.wfd, e.rfd, e.path = tmp, rfd, tmp.Name()
114
+
if out.ContentLength != nil { e.size = *out.ContentLength } else { e.size = -1 }
115
+
c.mu.Lock(); c.curBytes += max(e.size,0); c.evictLocked(key); c.mu.Unlock() // reserve budget
116
+
go c.pump(e, out.Body)
117
+
}
118
+
```
119
+
120
+
Note the `GetObject` runs **outside** `c.mu` (inside `once.Do`), so concurrent
121
+
opens of _different_ keys never serialize on the global lock; only the brief
122
+
header fetch for the _same_ key is serialized (subsequent openers see `once`
123
+
already done and attach instantly).
124
+
125
+
### 3. `pump`: write-through + advance watermark + trim RAM window
126
+
127
+
```go
128
+
func (c *PackCache) pump(e *packEntry, body io.ReadCloser) {
129
+
defer body.Close()
130
+
chunk := make([]byte, 256<<10)
131
+
for {
132
+
m, rerr := body.Read(chunk)
133
+
if m > 0 {
134
+
if _, werr := e.wfd.Write(chunk[:m]); werr != nil { e.fail(werr); return }
135
+
e.mu.Lock()
136
+
e.win = append(e.win, chunk[:m]...)
137
+
e.n += int64(m)
138
+
if int64(len(e.win)) > ringCap { // drop prefix now safe on disk
139
+
drop := int64(len(e.win)) - ringCap
140
+
e.win = e.win[drop:]; e.winStart += drop
141
+
}
142
+
e.cond.Broadcast()
143
+
e.mu.Unlock()
144
+
}
145
+
if rerr == io.EOF { break }
146
+
if rerr != nil { e.fail(rerr); return }
147
+
}
148
+
e.wfd.Close()
149
+
e.mu.Lock()
150
+
e.done = true; e.size = e.n; e.win = nil; e.winStart = e.n // free RAM; disk has it all
151
+
e.cond.Broadcast(); e.mu.Unlock()
152
+
// if evicted && refs==0 during streaming, close rfd + remove (orphan cleanup)
153
+
}
154
+
```
155
+
156
+
`e.fail(err)` sets `e.err`, broadcasts, closes `wfd`, removes the temp file, and
157
+
drops the entry from the map so a later open re-downloads (mirrors today's
158
+
failed-download cleanup). Write-through means **every byte `< n` is on disk**, so
159
+
`win` is purely a hot cache; trimming its prefix never loses data.
160
+
161
+
### 4. Reader: `packCachedFile` becomes a streaming view over the entry
162
+
163
+
Keep the type name `packCachedFile` (minimizes churn in `basic.go` and the
164
+
existing tests, which call `.Read`/`.ReadAt`/`.Seek`/`.Name`/`.Close`). It no
165
+
longer embeds `*os.File`; it holds `*packEntry` + its own cursor:
166
+
167
+
```go
168
+
type packCachedFile struct {
169
+
e *packEntry
170
+
name string
171
+
pos int64
172
+
closed bool
173
+
}
174
+
```
175
+
176
+
Core read primitive — `readAt` blocks only until the requested range is
177
+
available:
178
+
179
+
```go
180
+
func (f *packCachedFile) ReadAt(p []byte, off int64) (int, error) {
181
+
if f.closed { return 0, os.ErrClosed } // FSObject reopen contract (see CLAUDE.md)
182
+
return f.e.readAt(p, off)
183
+
}
184
+
185
+
func (e *packEntry) readAt(p []byte, off int64) (int, error) {
186
+
if off < 0 { return 0, errNegativeOffset }
187
+
e.mu.Lock()
188
+
for {
189
+
if e.err != nil { err := e.err; e.mu.Unlock(); return 0, err }
190
+
if e.size >= 0 && off >= e.size { e.mu.Unlock(); return 0, io.EOF }
191
+
if off+int64(len(p)) <= e.n { // full range available
192
+
// serve from RAM window if covered, else from disk fd
193
+
if off >= e.winStart {
194
+
n := copy(p, e.win[off-e.winStart:])
195
+
e.mu.Unlock(); return n, nil
196
+
}
197
+
rfd := e.rfd; e.mu.Unlock()
198
+
return rfd.ReadAt(p, off) // below window → disk (page cache)
199
+
}
200
+
if e.done { // range extends past EOF
201
+
rfd := e.rfd; e.mu.Unlock()
202
+
return rfd.ReadAt(p, off) // returns (partial, io.EOF) correctly
203
+
}
204
+
e.cond.Wait() // ahead of watermark → wait for more
205
+
}
206
+
}
207
+
```
208
+
209
+
`Read`/`Seek` reuse `readAt` against `f.pos` (Seek tracks pos; `SeekEnd` waits
210
+
for `size` to be known — it is set at header time, so no real wait). `Write*`,
211
+
`Truncate`, `Lock`/`Unlock`, `Name`, `Stat` mirror the current
212
+
`packCachedFile`; `Stat().Size()` returns `e.size`.
213
+
214
+
`Close`:
215
+
216
+
```go
217
+
func (f *packCachedFile) Close() error {
218
+
if f.closed { return ErrFileClosed-or-nil-per-current }
219
+
f.closed = true
220
+
f.e.cache.release(f.e) // refs--; if evicted && refs==0 → close rfd, (wfd already closed)
221
+
return nil
222
+
}
223
+
```
224
+
225
+
### 5. Eviction with refcount + unlink-while-open
226
+
227
+
`evictLocked` picks the least-recently-used victim (skipping `keep` and entries
228
+
with no `path` yet), then:
229
+
230
+
- `os.Remove(victim.path)` — unlink; live `wfd`/`rfd`/reader fds keep working on
231
+
the inode (Linux).
232
+
- `victim.evicted = true`, `curBytes -= victim.size`, delete from `entries` map
233
+
(new opens re-download).
234
+
- If `victim.refs == 0 && victim.done`, close `rfd` now; otherwise the last
235
+
`release` (or `pump` finishing) closes it. The shared `rfd` is what makes
236
+
unlink-while-open work for readers that haven't issued a disk read yet —
237
+
there's no per-reader `os.Open(path)` that could race the unlink.
238
+
239
+
`Cleanup` additionally closes any open `wfd`/`rfd` before `os.RemoveAll(dir)`.
240
+
241
+
## Files to modify
242
+
243
+
- `internal/s3fs/packcache.go` — the whole rewrite above (entry, `open`,
244
+
`start`, `pump`, `readAt`, reader methods, `evictLocked`, refcount/`release`,
245
+
`Cleanup`). Add `ringCap` const and `errNegativeOffset`.
246
+
- `internal/s3fs/packcache_test.go` — keep existing tests passing (they use a
247
+
synchronous full-bytes body, which still works: the pump drains it before the
248
+
first read in practice, and reads block-then-serve regardless). Add a
249
+
controllable body to exercise the streaming path (see Verification).
250
+
- No changes needed in `basic.go`/`chroot.go`/`filesystem.go`/`main.go`:
251
+
`open` still returns a `billy.File`; `WithPackCache`, flags, and Chroot
252
+
sharing are untouched.
253
+
254
+
Reuse: the read-while-write blocking pattern is the same idea already proven in
255
+
`internal/s3fs/tempfs.go` (`tempBuffer.readAt` returns `io.EOF` past the end for
256
+
the _write_ path); here the read path **blocks on a cond** instead of returning
257
+
EOF, because go-git's `packfile.FSObject`/scanner on the clone path expects
258
+
`ReadAt` to fill its buffer, not retry. The `observeS3("GetObject", …)` metric
259
+
hook stays at the header fetch. The FSObject `os.ErrClosed` contract
260
+
(`internal/s3fs/file.go`, documented in CLAUDE.md) is preserved by the `closed`
261
+
guard returning `os.ErrClosed` from `Read`/`ReadAt`/`Seek`.
262
+
263
+
## Verification
264
+
265
+
1. `go test ./internal/s3fs/...` — existing `TestPackCache*` must stay green
266
+
(downloads-once, independent cursors, closed-read-fails, missing-object,
267
+
eviction, `isPackCacheable`).
268
+
2. New table-driven tests (follow `xe-go:go-table-driven-tests`) in
269
+
`packcache_test.go` using a **gated body** — an `io.ReadCloser` that releases
270
+
bytes only when the test signals (channel/`sync.Cond`), wrapping the
271
+
`packStub` pattern:
272
+
- **serve-while-downloading:** open returns before the body is fully
273
+
released; `ReadAt` at a low offset returns once that prefix is released,
274
+
_before_ the tail is; assert it does not wait for EOF.
275
+
- **read ahead of watermark blocks then unblocks:** `ReadAt` for an offset
276
+
past the released watermark blocks; releasing more bytes unblocks it with
277
+
correct data (run the read in a goroutine, assert it's pending, release,
278
+
assert completion).
279
+
- **below-window reads hit disk:** with `ringCap` set small in the test,
280
+
release > ringCap bytes, then `ReadAt` at offset 0 (now trimmed from RAM)
281
+
returns correct bytes from the disk fd.
282
+
- **error mid-stream:** body returns a non-EOF error after N bytes; in-flight
283
+
and subsequent `ReadAt`s past N return that error; a later open re-downloads.
284
+
- **header not-found:** unchanged — `open` returns `fs.ErrNotExist`
285
+
synchronously (no goroutine launched).
286
+
- **eviction unlink-while-open:** existing test semantics, plus a variant
287
+
evicting an entry that is _still streaming_ and confirming the holding
288
+
reader finishes correctly.
289
+
- **os.ErrClosed after Close** during streaming.
290
+
- **concurrent readers** of one streaming key all see identical full bytes;
291
+
`GetObject` count == 1.
292
+
3. End-to-end clone against a real bucket (manual): build
293
+
`go build -o objgitd ./cmd/objgitd`, run with `-pack-cache-bytes` set, and
294
+
`git clone http://localhost:8080/<repo>.git` of a repo with a large pack;
295
+
confirm the clone makes progress (objects counting) while the pack is still
296
+
downloading rather than stalling until it completes, and that a second
297
+
concurrent clone of the same repo benefits from the shared in-flight entry.
298
+
Watch `objgit_s3_requests_total{operation="GetObject"}` stays ~one per
299
+
pack-dir file.
300
+
4. `go test ./cmd/objgitd/...` — protocol tests still pass (clone/fetch over
301
+
HTTP, git://, SSH) with `git` on PATH.
+135
docs/plans/s3fs-reduce-listobjectsv2.md
+135
docs/plans/s3fs-reduce-listobjectsv2.md
···
1
+
# Reduce ListObjectsV2 volume in objgitd
2
+
3
+
## Context
4
+
5
+
objgitd issues a high volume of `ListObjectsV2` calls against Tigris. The
6
+
codebase already has a sophisticated listing cache (`internal/s3fs/listingcache.go`)
7
+
with a **subtree-scan** optimization meant to collapse per-folder listings into a
8
+
single recursive scan — but investigation shows that optimization is **effectively
9
+
dead in any real deployment**, and a background warmer multiplies the remaining
10
+
waste. This plan fixes the root cause so the subtree optimization actually
11
+
engages.
12
+
13
+
## Root-cause findings (from reading the code)
14
+
15
+
**1. Subtree caching never matches chrooted repos — the big one.**
16
+
Every repo is served through a `Chroot`: both `loadOrInit` (`cmd/objgitd/git_protocol.go:192`)
17
+
and go-git's `FilesystemLoader.Load` (`plumbing/transport/loader.go:46`) chroot the
18
+
filesystem to the repo path. So canonical S3 keys are `<repo>.git/refs/...` and
19
+
`<repo>.git/objects/...`.
20
+
21
+
But the recursive-root match is anchored at the bucket root:
22
+
23
+
```go
24
+
// listingcache.go:290
25
+
func (c *ListingCache) recursiveRoot(prefix string) (string, bool) {
26
+
for _, r := range c.roots { // default {"refs/"}
27
+
if prefix == r || strings.HasPrefix(prefix, r) { ... }
28
+
}
29
+
}
30
+
```
31
+
32
+
`strings.HasPrefix("myrepo.git/refs/heads/", "refs/")` is **false**. So the
33
+
default `-s3-cache-recursive-prefixes refs/` matches nothing once a repo lives at
34
+
`myrepo.git/…`, and **every** listing falls to delimited per-folder mode
35
+
(`listFolder`).
36
+
37
+
**2. Loose-object negative lookups storm `objects/<xx>/`.**
38
+
go-git runs non-exclusive, so `hasObject` returns nil without listing
39
+
(`dotgit.go:646`) and `Object`/`ObjectStat` go straight to
40
+
`fs.Open`/`fs.Stat("objects/<xx>/<38hex>")` for any object not already in the
41
+
in-memory LRU (`dotgit.go:753,771`). Because objgitd keeps packs whole (no loose
42
+
objects), each distinct two-hex prefix probed → `resolve` → `cache.list("…/objects/<xx>/")`
43
+
→ one delimited `ListObjectsV2` returning empty. Up to **256 list calls per clone**,
44
+
none of which the (dead) subtree cache can absorb.
45
+
46
+
**3. The warmer multiplies it.**
47
+
`list()` calls `touch(prefix)` for every listed prefix, so `seen` accumulates all
48
+
~256 `objects/<xx>/` prefixes plus the refs folders. With `-s3-cache-refresh 30s`
49
+
and `-s3-cache-idle 10m`, `RunWarmer` re-lists every seen prefix every 30s for up
50
+
to 10 minutes — on the order of **thousands of background `ListObjectsV2` calls
51
+
following a single clone**, even with no further traffic. (`warmOnce` _does_
52
+
dedupe recursive prefixes to their root — but since no prefix matches a recursive
53
+
root today, nothing is deduped.)
54
+
55
+
## Fix: register each chroot repo root as a recursive subtree root
56
+
57
+
The architecture already gives us the right unit: **a chroot == a repo**. Make
58
+
each `Chroot` register its own root with the listing cache as a recursive subtree
59
+
prefix. Then one delimiter-less `ListObjectsV2` over `myrepo.git/` serves the
60
+
entire repo (refs, `objects/pack/` listing, HEAD, config, packed-refs, _and_ every
61
+
loose-object negative lookup) from a single scan — for objgitd's whole-pack repos
62
+
that subtree is a dozen-ish objects.
63
+
64
+
This single change:
65
+
66
+
- revives the subtree optimization (currently dead) for the real chrooted layout;
67
+
- extends it to `objects/`, eliminating the ~256 negative-lookup list calls/clone;
68
+
- collapses the warmer from ~260 lists/tick/repo to **1** (warmOnce already dedupes
69
+
to the root).
70
+
71
+
`MaxSubtreeKeys` (default 50000) still guards pathological repos: an oversized
72
+
subtree is marked `Truncated` and the code falls back to delimited per-folder
73
+
listing automatically (`list()` at `listingcache.go:206`).
74
+
75
+
### Changes
76
+
77
+
**`internal/s3fs/listingcache.go`** — make `roots` mutable and add a registrar:
78
+
79
+
- Change `roots []string` to a concurrency-safe snapshot. Simplest:
80
+
`roots atomic.Pointer[[]string]` (kept longest-first), or a `sync.RWMutex`
81
+
guarding the slice. `recursiveRoot` is hot and read-only, so an atomic snapshot
82
+
read is preferred.
83
+
- Add `func (c *ListingCache) registerRoot(root string)`: normalize to a trailing
84
+
`/`, no-op if already present or if `c == nil`, otherwise insert and re-sort
85
+
longest-first (reuse the logic in `normalizeRoots`, `listingcache.go:331`).
86
+
- `recursiveRoot` reads the snapshot instead of the immutable field. No logic
87
+
change — once `myrepo.git/` is a root it returns it for any prefix beneath it.
88
+
89
+
**`internal/s3fs/chroot.go`** — register the new root on chroot:
90
+
91
+
- After building `nfs` (`chroot.go:21`), if `fs3.cache != nil && p != ""` call
92
+
`fs3.cache.registerRoot(p + "/")`. `p` is the joined canonical root
93
+
(e.g. `myrepo.git`); the cache wants prefixes ending in `/`.
94
+
95
+
No change needed in `cmd/objgitd` — both load paths chroot already, so registration
96
+
happens automatically the first time a repo's storer is built.
97
+
98
+
### Notes / decisions
99
+
100
+
- **Keep the `-s3-cache-recursive-prefixes refs/` default.** It is now redundant
101
+
for chrooted repos but harmless, and still works for a bucket-root single-repo
102
+
layout or as an explicit override.
103
+
- **Whole-repo root vs. per-subdir (`objects/`+`refs/`).** Whole-repo is simpler
104
+
and strictly fewer scans (1 vs 2). Invalidation blast radius is fine: `invalidate`
105
+
already walks to the bucket root (`listingcache.go:472`), so any push/ref update
106
+
invalidates the repo subtree and the next read re-scans once.
107
+
- **Optional hardening (only if a giant-repo regression shows up):** have `warmOnce`
108
+
skip a root whose last subtree result was `Truncated`, so the warmer doesn't
109
+
re-scan up to `MaxSubtreeKeys` every tick for a pathological repo. Not part of the
110
+
core fix; objgitd's whole-pack repos never hit this.
111
+
112
+
## Verification
113
+
114
+
1. `go build ./...` and `go test ./internal/s3fs/...`.
115
+
2. **New table-driven test** in `internal/s3fs/listingcache_test.go` using the
116
+
existing `stubClient` (has an atomic `lists` counter and honors delimiter/
117
+
pagination, `listingcache_test.go:25-96`). Pattern after `TestListingCacheChrootShares`
118
+
(`listingcache_test.go:461`):
119
+
- Seed the stub with `myrepo.git/refs/heads/main`, `myrepo.git/objects/pack/pack-X.pack`,
120
+
`myrepo.git/objects/pack/pack-X.idx`, `myrepo.git/HEAD`.
121
+
- `Chroot("myrepo.git")`, then drive the loose-object access pattern: `Stat`
122
+
several non-existent `objects/<xx>/<hash>` paths across different `xx`,
123
+
`ReadDir("objects/pack")`, `ReadDir("refs/heads")`.
124
+
- Assert `stub.lists` is **1** (one subtree scan), versus the pre-fix behavior of
125
+
one list per distinct `objects/<xx>/` prefix + one per refs folder.
126
+
- Add a write (`Create`+`Rename` a new pack, or invalidate) and assert exactly
127
+
one additional scan on the next read (re-scan after invalidation).
128
+
- Use `xe-go:go-table-driven-tests` conventions (see MEMORY: tests here follow
129
+
that skill).
130
+
3. **Manual end-to-end** against a real bucket: run `./objgitd -bucket $BUCKET
131
+
-http-bind :8080 -allow-push`, push a repo, then `git clone` it twice. Watch
132
+
`objgit_s3_requests_total{api="ListObjectsV2"}` on `/metrics` (or the
133
+
listing-cache hit/miss log emitted from `main.go:111`): the per-clone
134
+
`ListObjectsV2` count should drop from order-of-hundreds to a small constant,
135
+
and the steady-state warmer rate should drop to ~1 per repo per refresh tick.
+443
-73
internal/s3fs/packcache.go
+443
-73
internal/s3fs/packcache.go
···
2
2
3
3
import (
4
4
"context"
5
+
"errors"
5
6
"fmt"
6
7
"io"
7
8
"io/fs"
···
11
12
"time"
12
13
13
14
"github.com/aws/aws-sdk-go-v2/service/s3"
15
+
"github.com/go-git/go-billy/v6"
14
16
)
15
17
18
+
// ringCap bounds the in-RAM trailing window held for an in-flight download.
19
+
// Every byte below the watermark is already written to the temp file, so the
20
+
// window is only a hot cache for the most-recently-downloaded bytes; reads that
21
+
// fall below it are served from the disk fd. Peak RAM is therefore roughly
22
+
// (concurrent in-flight packs) * ringCap regardless of pack size.
23
+
const ringCap = 4 << 20 // 4 MiB
24
+
25
+
var errNegativeOffset = errors.New("s3fs: negative offset")
26
+
16
27
// isPackCacheable reports whether key names an immutable pack-directory file
17
28
// that benefits from the local temp-file cache. These files are content-
18
29
// addressed (pack-<sha>.{pack,idx,rev}) and re-read with random access many
···
30
41
// fresh S3 GetObject and a clone never completes. The cache is shared by pointer
31
42
// across an S3FS and all of its Chroot children.
32
43
//
44
+
// Downloads stream: open launches a background pump that writes the S3 body to a
45
+
// temp file while advancing a watermark, and returns a reader immediately.
46
+
// Reads of already-downloaded ranges return at once; reads ahead of the
47
+
// watermark block only until that range arrives (the body is one sequential
48
+
// stream, so the watermark is monotonic). This overlaps the download with the
49
+
// clone instead of blocking it until the whole pack lands.
50
+
//
33
51
// Entries are keyed by S3 object key. Each downloads its object once and is
34
52
// reused across opens; a total-bytes budget evicts the least-recently-opened
35
-
// entries. Eviction unlinks the temp file, which on Linux leaves already-open
36
-
// readers working until they close, so eviction never corrupts an in-flight
37
-
// read.
53
+
// entries. Eviction unlinks the temp file, which on Linux leaves the in-flight
54
+
// writer and already-open readers working on the unlinked inode, so eviction
55
+
// never corrupts an in-flight read.
38
56
type PackCache struct {
39
57
dir string
40
58
maxBytes int64
···
45
63
seq uint64 // monotonic open counter; entry.used orders the LRU
46
64
}
47
65
48
-
// packEntry is one cached object. once guards the single download; path/size/err
49
-
// are set by it. used is the seq of the most recent open, for LRU ordering.
66
+
// packEntry is one cached object, possibly still downloading. once guards the
67
+
// header GetObject and the launch of the pump goroutine; the pump then fills the
68
+
// temp file and the RAM window concurrently with readers.
50
69
type packEntry struct {
51
-
key string
52
-
once sync.Once
70
+
cache *PackCache
71
+
key string
72
+
once sync.Once
73
+
74
+
mu sync.Mutex
75
+
cond *sync.Cond
76
+
77
+
wfd *os.File // write side: the pump appends sequentially
78
+
rfd *os.File // shared read side: serves offsets < n; survives unlink
53
79
path string
54
-
size int64
55
-
err error
56
-
used uint64
80
+
81
+
win []byte // trailing RAM window covering [winStart, n)
82
+
winStart int64
83
+
n int64 // bytes downloaded and written so far (monotonic watermark)
84
+
size int64 // total size from Content-Length; -1 until known
85
+
done bool // body fully drained successfully
86
+
err error // terminal error (header or body)
87
+
88
+
used uint64 // seq of the most recent open, for LRU ordering
89
+
refs int // live reader handles
90
+
reserved int64 // bytes this entry has added to PackCache.curBytes
91
+
evicted bool // path unlinked; close fds once refs hits 0
92
+
93
+
closeOnce sync.Once // guards closing rfd exactly once
57
94
}
58
95
59
96
// NewPackCache creates a pack cache writing temp files under a fresh directory
···
76
113
}, nil
77
114
}
78
115
79
-
// Cleanup removes the cache's temp directory and all files in it. Already-open
80
-
// readers keep working (unlinked-while-open); new opens after Cleanup fail.
116
+
// Cleanup removes the cache's temp directory and all files in it, closing any
117
+
// open read/write handles first. Readers that still hold a *packCachedFile keep
118
+
// their entry pointer but will see closed fds; new opens after Cleanup fail.
81
119
func (c *PackCache) Cleanup() error {
82
120
c.mu.Lock()
83
121
dir := c.dir
122
+
entries := c.entries
84
123
c.entries = map[string]*packEntry{}
85
124
c.curBytes = 0
86
125
c.mu.Unlock()
126
+
for _, e := range entries {
127
+
e.closeRead()
128
+
e.mu.Lock()
129
+
if e.wfd != nil {
130
+
e.wfd.Close()
131
+
}
132
+
e.mu.Unlock()
133
+
}
87
134
return os.RemoveAll(dir)
88
135
}
89
136
90
-
// open returns a billy.File for key, downloading the object to a temp file on
91
-
// first use and serving from that file thereafter. Each call returns an
92
-
// independent *os.File handle so concurrent readers have their own seek cursor.
93
-
func (c *PackCache) open(ctx context.Context, client s3Client, bucket, key, name string) (*packCachedFile, error) {
137
+
// open returns a billy.File for key. On first use it fetches the object header
138
+
// (so not-found and auth errors surface synchronously) and launches a pump that
139
+
// streams the body to a temp file; the returned reader serves bytes as they
140
+
// arrive without waiting for the whole download. Each call returns an
141
+
// independent handle with its own cursor over the shared entry.
142
+
func (c *PackCache) open(ctx context.Context, client s3Client, bucket, key, name string) (billy.File, error) {
94
143
c.mu.Lock()
95
144
e := c.entries[key]
96
145
if e == nil {
97
-
e = &packEntry{key: key}
146
+
e = &packEntry{cache: c, key: key, size: -1}
147
+
e.cond = sync.NewCond(&e.mu)
98
148
c.entries[key] = e
99
149
}
150
+
// Reserve a reference before releasing the lock so a concurrent open of
151
+
// another key can't evict-and-close this entry out from under us. Eviction
152
+
// may still unlink it (unlink-while-open), but won't close its fds while
153
+
// refs > 0.
154
+
e.refs++
155
+
c.seq++
156
+
e.used = c.seq
100
157
c.mu.Unlock()
101
158
102
-
e.once.Do(func() {
103
-
path, size, err := c.download(ctx, client, bucket, key)
104
-
if err != nil {
105
-
e.err = err
106
-
// Drop the failed entry so a later open retries the download.
107
-
c.mu.Lock()
108
-
if c.entries[key] == e {
109
-
delete(c.entries, key)
110
-
}
111
-
c.mu.Unlock()
112
-
return
159
+
e.once.Do(func() { c.start(ctx, client, bucket, key, e) })
160
+
161
+
e.mu.Lock()
162
+
startErr := e.err
163
+
e.mu.Unlock()
164
+
if startErr != nil {
165
+
c.release(e)
166
+
return nil, startErr
167
+
}
168
+
169
+
return &packCachedFile{e: e, name: name}, nil
170
+
}
171
+
172
+
// start fetches the object and launches the background pump. It runs once per
173
+
// entry (guarded by entry.once) and outside c.mu, so concurrent opens of
174
+
// different keys don't serialise on the network call. A header/GetObject error
175
+
// is recorded on the entry and the entry is dropped so a later open retries.
176
+
func (c *PackCache) start(ctx context.Context, client s3Client, bucket, key string, e *packEntry) {
177
+
began := time.Now()
178
+
out, err := client.GetObject(ctx, &s3.GetObjectInput{Bucket: &bucket, Key: &key})
179
+
observeS3("GetObject", began, err)
180
+
if err != nil {
181
+
if isNotFound(err) {
182
+
err = &os.PathError{Op: "open", Path: key, Err: fs.ErrNotExist}
183
+
} else {
184
+
err = fmt.Errorf("pack cache GetObject %q: %w", key, err)
113
185
}
114
-
e.path, e.size = path, size
186
+
c.dropFailed(e, err)
187
+
return
188
+
}
189
+
190
+
tmp, terr := os.CreateTemp(c.dir, "obj-")
191
+
if terr != nil {
192
+
out.Body.Close()
193
+
c.dropFailed(e, fmt.Errorf("pack cache temp file: %w", terr))
194
+
return
195
+
}
196
+
rfd, rerr := os.Open(tmp.Name())
197
+
if rerr != nil {
198
+
tmp.Close()
199
+
os.Remove(tmp.Name())
200
+
out.Body.Close()
201
+
c.dropFailed(e, fmt.Errorf("pack cache reopen %q: %w", tmp.Name(), rerr))
202
+
return
203
+
}
204
+
205
+
size := int64(-1)
206
+
if out.ContentLength != nil {
207
+
size = *out.ContentLength
208
+
}
209
+
210
+
e.mu.Lock()
211
+
e.wfd, e.rfd, e.path, e.size = tmp, rfd, tmp.Name(), size
212
+
e.mu.Unlock()
213
+
214
+
if size > 0 {
115
215
c.mu.Lock()
216
+
e.reserved = size
116
217
c.curBytes += size
117
218
c.evictLocked(key)
118
219
c.mu.Unlock()
119
-
})
120
-
if e.err != nil {
121
-
return nil, e.err
220
+
}
221
+
222
+
go c.pump(e, out.Body)
223
+
}
224
+
225
+
// dropFailed records a terminal error on the entry, wakes any waiters, and
226
+
// removes it from the map so a later open re-fetches.
227
+
func (c *PackCache) dropFailed(e *packEntry, err error) {
228
+
e.mu.Lock()
229
+
e.err = err
230
+
e.cond.Broadcast()
231
+
e.mu.Unlock()
232
+
c.mu.Lock()
233
+
if c.entries[e.key] == e {
234
+
delete(c.entries, e.key)
122
235
}
236
+
c.mu.Unlock()
237
+
}
123
238
124
-
f, err := os.Open(e.path)
125
-
if err != nil {
126
-
// The cached file was evicted/cleaned between download and open; retry
127
-
// through a fresh entry so the object is fetched again.
128
-
c.mu.Lock()
129
-
if c.entries[key] == e {
130
-
delete(c.entries, key)
239
+
// pump streams body into the entry's temp file, advancing the watermark and the
240
+
// trailing RAM window as bytes arrive, and broadcasting so blocked readers wake.
241
+
func (c *PackCache) pump(e *packEntry, body io.ReadCloser) {
242
+
defer body.Close()
243
+
chunk := make([]byte, 256<<10)
244
+
for {
245
+
m, rerr := body.Read(chunk)
246
+
if m > 0 {
247
+
if _, werr := e.wfd.Write(chunk[:m]); werr != nil {
248
+
c.failStream(e, fmt.Errorf("pack cache write %q: %w", e.key, werr))
249
+
return
250
+
}
251
+
e.mu.Lock()
252
+
e.win = append(e.win, chunk[:m]...)
253
+
e.n += int64(m)
254
+
if int64(len(e.win)) > ringCap {
255
+
drop := int64(len(e.win)) - ringCap
256
+
e.win = e.win[drop:]
257
+
e.winStart += drop
258
+
}
259
+
e.cond.Broadcast()
260
+
e.mu.Unlock()
131
261
}
132
-
c.mu.Unlock()
133
-
return nil, err
262
+
if rerr == io.EOF {
263
+
break
264
+
}
265
+
if rerr != nil {
266
+
c.failStream(e, fmt.Errorf("pack cache download %q: %w", e.key, rerr))
267
+
return
268
+
}
134
269
}
135
270
271
+
if cerr := e.wfd.Close(); cerr != nil {
272
+
c.failStream(e, fmt.Errorf("pack cache flush %q: %w", e.key, cerr))
273
+
return
274
+
}
275
+
276
+
e.mu.Lock()
277
+
e.wfd = nil
278
+
e.done = true
279
+
e.size = e.n
280
+
e.win = nil // every byte is on disk now; drop the RAM window
281
+
e.winStart = e.n
282
+
e.cond.Broadcast()
283
+
e.mu.Unlock()
284
+
285
+
// If we under-reserved (size was unknown), reconcile the budget now.
136
286
c.mu.Lock()
137
-
c.seq++
138
-
e.used = c.seq
287
+
if e.reserved != e.n {
288
+
c.curBytes += e.n - e.reserved
289
+
e.reserved = e.n
290
+
c.evictLocked(e.key)
291
+
}
139
292
c.mu.Unlock()
140
293
141
-
return &packCachedFile{File: f, name: name}, nil
294
+
c.maybeCloseEvicted(e)
142
295
}
143
296
144
-
// download streams the full object to a temp file and returns its path and size.
145
-
func (c *PackCache) download(ctx context.Context, client s3Client, bucket, key string) (string, int64, error) {
146
-
start := time.Now()
147
-
out, err := client.GetObject(ctx, &s3.GetObjectInput{Bucket: &bucket, Key: &key})
148
-
observeS3("GetObject", start, err)
149
-
if err != nil {
150
-
if isNotFound(err) {
151
-
return "", 0, &os.PathError{Op: "open", Path: key, Err: fs.ErrNotExist}
152
-
}
153
-
return "", 0, fmt.Errorf("pack cache GetObject %q: %w", key, err)
297
+
// failStream records a mid-stream download error, discards the partial temp
298
+
// file, and drops the entry. Already-open readers see the error on their next
299
+
// blocked or out-of-range read.
300
+
func (c *PackCache) failStream(e *packEntry, err error) {
301
+
e.mu.Lock()
302
+
if e.wfd != nil {
303
+
e.wfd.Close()
304
+
e.wfd = nil
154
305
}
155
-
defer out.Body.Close()
306
+
path := e.path
307
+
reserved := e.reserved
308
+
e.reserved = 0
309
+
e.err = err
310
+
e.cond.Broadcast()
311
+
e.mu.Unlock()
312
+
313
+
e.closeRead()
314
+
if path != "" {
315
+
os.Remove(path)
316
+
}
156
317
157
-
tmp, err := os.CreateTemp(c.dir, "obj-")
158
-
if err != nil {
159
-
return "", 0, fmt.Errorf("pack cache temp file: %w", err)
318
+
c.mu.Lock()
319
+
c.curBytes -= reserved
320
+
if c.entries[e.key] == e {
321
+
delete(c.entries, e.key)
160
322
}
161
-
n, err := io.Copy(tmp, out.Body)
162
-
if cerr := tmp.Close(); err == nil {
163
-
err = cerr
323
+
c.mu.Unlock()
324
+
}
325
+
326
+
// release drops one reader reference. When the last reader of an evicted entry
327
+
// closes, its shared read fd is closed (the temp file is already unlinked).
328
+
func (c *PackCache) release(e *packEntry) {
329
+
c.mu.Lock()
330
+
e.refs--
331
+
closeNow := e.refs == 0 && e.evicted
332
+
c.mu.Unlock()
333
+
if closeNow {
334
+
e.closeRead()
164
335
}
165
-
if err != nil {
166
-
os.Remove(tmp.Name())
167
-
return "", 0, fmt.Errorf("pack cache download %q: %w", key, err)
336
+
}
337
+
338
+
// maybeCloseEvicted closes an evicted entry's read fd if no readers remain.
339
+
// Called after the pump finishes, since an entry can be evicted mid-stream.
340
+
func (c *PackCache) maybeCloseEvicted(e *packEntry) {
341
+
c.mu.Lock()
342
+
closeNow := e.refs == 0 && e.evicted
343
+
c.mu.Unlock()
344
+
if closeNow {
345
+
e.closeRead()
168
346
}
169
-
return tmp.Name(), n, nil
347
+
}
348
+
349
+
// closeRead closes the shared read fd exactly once.
350
+
func (e *packEntry) closeRead() {
351
+
e.closeOnce.Do(func() {
352
+
e.mu.Lock()
353
+
rfd := e.rfd
354
+
e.mu.Unlock()
355
+
if rfd != nil {
356
+
rfd.Close()
357
+
}
358
+
})
170
359
}
171
360
172
361
// evictLocked removes least-recently-opened entries until the cache is within
173
362
// budget. keep is never evicted (it is the entry the caller just populated).
174
-
// Callers hold c.mu. A non-positive maxBytes disables eviction.
363
+
// Callers hold c.mu. A non-positive maxBytes disables eviction. Eviction unlinks
364
+
// the temp file even while readers or the in-flight writer hold it open; they
365
+
// survive via their fds on the unlinked inode (Linux). The shared read fd is
366
+
// closed only once the last reader leaves (see release/maybeCloseEvicted).
175
367
func (c *PackCache) evictLocked(keep string) {
176
368
if c.maxBytes <= 0 {
177
369
return
···
189
381
if victim == nil {
190
382
return // nothing evictable
191
383
}
192
-
os.Remove(victim.path) // open readers survive on Linux (unlinked fd)
193
-
c.curBytes -= victim.size
384
+
os.Remove(victim.path) // open fds survive on Linux (unlinked inode)
385
+
victim.mu.Lock()
386
+
victim.evicted = true
387
+
refs := victim.refs
388
+
victim.mu.Unlock()
389
+
c.curBytes -= victim.reserved
194
390
delete(c.entries, victim.key)
391
+
if refs == 0 {
392
+
victim.closeRead()
393
+
}
195
394
}
196
395
}
197
396
198
-
// packCachedFile is a read-only billy.File backed by a local temp file. It
199
-
// embeds *os.File for Read/ReadAt/Seek/Close/Stat and supplies the billy-only
200
-
// Lock/Unlock; writes are rejected.
397
+
// readAt fills p from the entry, blocking until the full requested range has
398
+
// been downloaded (or the download finishes or fails). This matches ReadAt's
399
+
// fill-or-error contract, which go-git's packfile.FSObject relies on. Bytes in
400
+
// the RAM window are served from memory; bytes that have scrolled below the
401
+
// window are served from the shared read fd.
402
+
func (e *packEntry) readAt(p []byte, off int64) (int, error) {
403
+
if off < 0 {
404
+
return 0, errNegativeOffset
405
+
}
406
+
if len(p) == 0 {
407
+
return 0, nil
408
+
}
409
+
e.mu.Lock()
410
+
for {
411
+
if e.err != nil {
412
+
err := e.err
413
+
e.mu.Unlock()
414
+
return 0, err
415
+
}
416
+
if e.size >= 0 && off >= e.size {
417
+
e.mu.Unlock()
418
+
return 0, io.EOF
419
+
}
420
+
if off+int64(len(p)) <= e.n {
421
+
if off >= e.winStart {
422
+
n := copy(p, e.win[off-e.winStart:])
423
+
e.mu.Unlock()
424
+
return n, nil
425
+
}
426
+
rfd := e.rfd
427
+
e.mu.Unlock()
428
+
return rfd.ReadAt(p, off)
429
+
}
430
+
if e.done {
431
+
// Range extends past EOF: serve what exists from disk; os.File.ReadAt
432
+
// returns the partial read plus io.EOF.
433
+
rfd := e.rfd
434
+
e.mu.Unlock()
435
+
return rfd.ReadAt(p, off)
436
+
}
437
+
e.cond.Wait()
438
+
}
439
+
}
440
+
441
+
// readSome fills p with whatever is available at off (at least one byte once any
442
+
// data past off exists), without waiting for the full buffer. It backs the
443
+
// sequential Read path, whose callers (io.ReadAll, io.Copy) tolerate short
444
+
// reads.
445
+
func (e *packEntry) readSome(p []byte, off int64) (int, error) {
446
+
if off < 0 {
447
+
return 0, errNegativeOffset
448
+
}
449
+
if len(p) == 0 {
450
+
return 0, nil
451
+
}
452
+
e.mu.Lock()
453
+
for {
454
+
if e.err != nil {
455
+
err := e.err
456
+
e.mu.Unlock()
457
+
return 0, err
458
+
}
459
+
if e.size >= 0 && off >= e.size {
460
+
e.mu.Unlock()
461
+
return 0, io.EOF
462
+
}
463
+
if off < e.n {
464
+
if off >= e.winStart {
465
+
n := copy(p, e.win[off-e.winStart:])
466
+
e.mu.Unlock()
467
+
return n, nil
468
+
}
469
+
rfd := e.rfd
470
+
avail := e.n - off
471
+
e.mu.Unlock()
472
+
if int64(len(p)) > avail {
473
+
p = p[:avail]
474
+
}
475
+
return rfd.ReadAt(p, off)
476
+
}
477
+
if e.done { // off >= n and size unknown path: nothing more is coming
478
+
e.mu.Unlock()
479
+
return 0, io.EOF
480
+
}
481
+
e.cond.Wait()
482
+
}
483
+
}
484
+
485
+
// sizeOrWait returns the total size, waiting for the download to finish if the
486
+
// size was not advertised in the object header (Content-Length absent).
487
+
func (e *packEntry) sizeOrWait() (int64, error) {
488
+
e.mu.Lock()
489
+
defer e.mu.Unlock()
490
+
for {
491
+
if e.size >= 0 {
492
+
return e.size, nil
493
+
}
494
+
if e.err != nil {
495
+
return 0, e.err
496
+
}
497
+
if e.done {
498
+
return e.n, nil
499
+
}
500
+
e.cond.Wait()
501
+
}
502
+
}
503
+
504
+
// packCachedFile is a read-only billy.File backed by a (possibly still
505
+
// downloading) pack cache entry. Each handle carries its own cursor; ReadAt and
506
+
// the shared read fd are offset-explicit, so concurrent handles never interfere.
201
507
type packCachedFile struct {
202
-
*os.File
203
-
name string
508
+
e *packEntry
509
+
name string
510
+
pos int64
511
+
closed bool
204
512
}
205
513
206
514
func (f *packCachedFile) Name() string { return f.name }
207
515
516
+
func (f *packCachedFile) Read(p []byte) (int, error) {
517
+
if f.closed {
518
+
return 0, os.ErrClosed
519
+
}
520
+
n, err := f.e.readSome(p, f.pos)
521
+
f.pos += int64(n)
522
+
return n, err
523
+
}
524
+
525
+
func (f *packCachedFile) ReadAt(p []byte, off int64) (int, error) {
526
+
if f.closed {
527
+
return 0, os.ErrClosed
528
+
}
529
+
return f.e.readAt(p, off)
530
+
}
531
+
532
+
func (f *packCachedFile) Seek(offset int64, whence int) (int64, error) {
533
+
if f.closed {
534
+
return 0, os.ErrClosed
535
+
}
536
+
switch whence {
537
+
case io.SeekStart:
538
+
f.pos = offset
539
+
case io.SeekCurrent:
540
+
f.pos += offset
541
+
case io.SeekEnd:
542
+
size, err := f.e.sizeOrWait()
543
+
if err != nil {
544
+
return 0, err
545
+
}
546
+
f.pos = size + offset
547
+
default:
548
+
return 0, fmt.Errorf("s3fs: invalid whence %d", whence)
549
+
}
550
+
return f.pos, nil
551
+
}
552
+
553
+
func (f *packCachedFile) Stat() (fs.FileInfo, error) {
554
+
if f.closed {
555
+
return nil, os.ErrClosed
556
+
}
557
+
f.e.mu.Lock()
558
+
size := f.e.size
559
+
if size < 0 {
560
+
size = f.e.n
561
+
}
562
+
f.e.mu.Unlock()
563
+
return newFileInfo(f.name, size, time.Now()), nil
564
+
}
565
+
566
+
func (f *packCachedFile) Close() error {
567
+
if f.closed {
568
+
return ErrFileClosed
569
+
}
570
+
f.closed = true
571
+
f.e.cache.release(f.e)
572
+
return nil
573
+
}
574
+
208
575
func (f *packCachedFile) Write(p []byte) (int, error) { return 0, ErrCantWriteToReadOnly }
209
576
func (f *packCachedFile) WriteAt(p []byte, off int64) (int, error) { return 0, ErrCantWriteToReadOnly }
210
577
func (f *packCachedFile) Truncate(size int64) error { return ErrTruncateNotSupported }
211
578
func (f *packCachedFile) Lock() error { return ErrLockNotSupported }
212
579
func (f *packCachedFile) Unlock() error { return ErrLockNotSupported }
580
+
581
+
// Compile-time assertion: the streaming handle satisfies billy.File.
582
+
var _ billy.File = (*packCachedFile)(nil)
+404
-5
internal/s3fs/packcache_test.go
+404
-5
internal/s3fs/packcache_test.go
···
9
9
"os"
10
10
"sync"
11
11
"testing"
12
+
"time"
12
13
13
14
"github.com/aws/aws-sdk-go-v2/aws"
14
15
"github.com/aws/aws-sdk-go-v2/service/s3"
16
+
"github.com/go-git/go-billy/v6"
15
17
)
16
18
17
19
// packStub serves fixed object bytes and counts GetObject calls per key, so a
···
49
51
}, nil
50
52
}
51
53
54
+
// gateReader is an io.ReadCloser whose bytes are released to readers only when
55
+
// the test calls release/releaseAll, so a test can drive the streaming pump byte
56
+
// by byte and observe reads that race the download. Optionally it injects an
57
+
// error once the reader reaches a given offset.
58
+
type gateReader struct {
59
+
mu sync.Mutex
60
+
cond *sync.Cond
61
+
data []byte
62
+
released int
63
+
pos int
64
+
errAt int // -1 disables; otherwise return failErr once pos reaches it
65
+
failErr error
66
+
closed bool
67
+
}
68
+
69
+
func newGateReader(data []byte) *gateReader {
70
+
g := &gateReader{data: data, errAt: -1}
71
+
g.cond = sync.NewCond(&g.mu)
72
+
return g
73
+
}
74
+
75
+
func (g *gateReader) release(n int) {
76
+
g.mu.Lock()
77
+
g.released += n
78
+
if g.released > len(g.data) {
79
+
g.released = len(g.data)
80
+
}
81
+
g.cond.Broadcast()
82
+
g.mu.Unlock()
83
+
}
84
+
85
+
func (g *gateReader) releaseAll() { g.release(len(g.data)) }
86
+
87
+
func (g *gateReader) Read(p []byte) (int, error) {
88
+
g.mu.Lock()
89
+
defer g.mu.Unlock()
90
+
for {
91
+
if g.errAt >= 0 && g.pos >= g.errAt {
92
+
return 0, g.failErr
93
+
}
94
+
if g.pos < g.released {
95
+
n := copy(p, g.data[g.pos:g.released])
96
+
g.pos += n
97
+
return n, nil
98
+
}
99
+
if g.pos >= len(g.data) {
100
+
return 0, io.EOF
101
+
}
102
+
if g.closed {
103
+
return 0, io.ErrClosedPipe
104
+
}
105
+
g.cond.Wait()
106
+
}
107
+
}
108
+
109
+
func (g *gateReader) Close() error {
110
+
g.mu.Lock()
111
+
g.closed = true
112
+
g.cond.Broadcast()
113
+
g.mu.Unlock()
114
+
return nil
115
+
}
116
+
117
+
// gateStub serves a gateReader body per key so streaming behaviour can be
118
+
// driven deterministically.
119
+
type gateStub struct {
120
+
s3Client
121
+
mu sync.Mutex
122
+
gates map[string]*gateReader
123
+
gets map[string]int
124
+
}
125
+
126
+
func newGateStub() *gateStub {
127
+
return &gateStub{gates: map[string]*gateReader{}, gets: map[string]int{}}
128
+
}
129
+
130
+
func (s *gateStub) put(key string, g *gateReader) {
131
+
s.mu.Lock()
132
+
s.gates[key] = g
133
+
s.mu.Unlock()
134
+
}
135
+
136
+
func (s *gateStub) getCount(key string) int {
137
+
s.mu.Lock()
138
+
defer s.mu.Unlock()
139
+
return s.gets[key]
140
+
}
141
+
142
+
func (s *gateStub) GetObject(_ context.Context, in *s3.GetObjectInput, _ ...func(*s3.Options)) (*s3.GetObjectOutput, error) {
143
+
s.mu.Lock()
144
+
defer s.mu.Unlock()
145
+
key := aws.ToString(in.Key)
146
+
g, ok := s.gates[key]
147
+
if !ok {
148
+
return nil, notFound()
149
+
}
150
+
s.gets[key]++
151
+
return &s3.GetObjectOutput{
152
+
Body: g,
153
+
ContentLength: aws.Int64(int64(len(g.data))),
154
+
}, nil
155
+
}
156
+
52
157
// newTestPackCache builds a PackCache under t.TempDir with the given budget and
53
158
// registers Cleanup.
54
159
func newTestPackCache(t *testing.T, maxBytes int64) *PackCache {
···
61
166
return pc
62
167
}
63
168
64
-
func mustReadAll(t *testing.T, f *packCachedFile) []byte {
169
+
func mustReadAll(t *testing.T, f billy.File) []byte {
65
170
t.Helper()
66
171
if _, err := f.Seek(0, io.SeekStart); err != nil {
67
172
t.Fatalf("seek: %v", err)
···
73
178
return b
74
179
}
75
180
181
+
// readAtChan runs f.ReadAt in a goroutine and returns a channel that yields the
182
+
// result, so a test can assert whether a read is pending or has completed
183
+
// without hanging the test on a buggy block.
184
+
type readAtResult struct {
185
+
n int
186
+
err error
187
+
p []byte
188
+
}
189
+
190
+
func readAtChan(f billy.File, length int, off int64) <-chan readAtResult {
191
+
ch := make(chan readAtResult, 1)
192
+
go func() {
193
+
p := make([]byte, length)
194
+
n, err := f.ReadAt(p, off)
195
+
ch <- readAtResult{n: n, err: err, p: p}
196
+
}()
197
+
return ch
198
+
}
199
+
76
200
func TestPackCacheReads(t *testing.T) {
77
201
ctx := context.Background()
78
202
const key = "repo.git/objects/pack/pack-abc.pack"
···
82
206
tests := []struct {
83
207
name string
84
208
// check exercises one read path and returns (got, wantSlice) to compare.
85
-
check func(t *testing.T, f *packCachedFile) (got, exp []byte)
209
+
check func(t *testing.T, f billy.File) (got, exp []byte)
86
210
}{
87
211
{
88
212
name: "sequential Read",
89
-
check: func(t *testing.T, f *packCachedFile) ([]byte, []byte) {
213
+
check: func(t *testing.T, f billy.File) ([]byte, []byte) {
90
214
return mustReadAll(t, f), want
91
215
},
92
216
},
93
217
{
94
218
name: "ReadAt mid-object",
95
-
check: func(t *testing.T, f *packCachedFile) ([]byte, []byte) {
219
+
check: func(t *testing.T, f billy.File) ([]byte, []byte) {
96
220
p := make([]byte, 100)
97
221
if _, err := f.ReadAt(p, 8000); err != nil && err != io.EOF {
98
222
t.Fatalf("ReadAt: %v", err)
···
151
275
}
152
276
153
277
// TestPackCacheIndependentCursors verifies two open handles over the same cached
154
-
// file have independent seek positions (each gets its own *os.File).
278
+
// file have independent seek positions.
155
279
func TestPackCacheIndependentCursors(t *testing.T) {
156
280
ctx := context.Background()
157
281
const key = "r.git/objects/pack/pack-y.idx"
···
237
361
t.Fatal(err)
238
362
}
239
363
defer fa.Close()
364
+
// Touch A so its download completes and it is fully accounted in the budget
365
+
// before B arrives.
366
+
if got := mustReadAll(t, fa); !bytes.Equal(got, dataA) {
367
+
t.Fatalf("A bytes mismatch")
368
+
}
240
369
241
370
// Open B: total (2000) exceeds 1500, so A is evicted from the cache.
242
371
fb, err := pc.open(ctx, stub, "bucket", keyB, "b")
243
372
if err != nil {
244
373
t.Fatal(err)
374
+
}
375
+
if got := mustReadAll(t, fb); !bytes.Equal(got, dataB) {
376
+
t.Fatalf("B bytes mismatch")
245
377
}
246
378
fb.Close()
247
379
···
255
387
if err != nil {
256
388
t.Fatal(err)
257
389
}
390
+
if got := mustReadAll(t, fa2); !bytes.Equal(got, dataA) {
391
+
t.Fatalf("A re-download bytes mismatch")
392
+
}
258
393
fa2.Close()
259
394
if n := stub.getCount(keyA); n != 2 {
260
395
t.Errorf("A GetObject count = %d, want 2 (re-downloaded after eviction)", n)
261
396
}
262
397
}
263
398
399
+
// TestPackCacheServeWhileDownloading confirms a read of an already-arrived
400
+
// prefix returns without waiting for the rest of the object to download.
401
+
func TestPackCacheServeWhileDownloading(t *testing.T) {
402
+
ctx := context.Background()
403
+
const key = "r.git/objects/pack/stream.pack"
404
+
data := bytes.Repeat([]byte("STREAMDATA"), 1000) // 10 KiB
405
+
g := newGateReader(data)
406
+
stub := newGateStub()
407
+
stub.put(key, g)
408
+
pc := newTestPackCache(t, 0)
409
+
410
+
f, err := pc.open(ctx, stub, "bucket", key, "n")
411
+
if err != nil {
412
+
t.Fatalf("open: %v", err)
413
+
}
414
+
defer f.Close()
415
+
416
+
// Release only the first 200 bytes; the rest of the object is still
417
+
// "downloading" (gated). A read of the prefix must succeed promptly.
418
+
g.release(200)
419
+
ch := readAtChan(f, 100, 0)
420
+
select {
421
+
case res := <-ch:
422
+
if res.err != nil {
423
+
t.Fatalf("ReadAt prefix: %v", res.err)
424
+
}
425
+
if !bytes.Equal(res.p, data[:100]) {
426
+
t.Fatalf("ReadAt prefix bytes mismatch")
427
+
}
428
+
case <-time.After(2 * time.Second):
429
+
t.Fatal("ReadAt of arrived prefix blocked waiting for full download")
430
+
}
431
+
432
+
// Let the rest finish so Close/Cleanup don't race a stuck pump.
433
+
g.releaseAll()
434
+
}
435
+
436
+
// TestPackCacheReadAheadBlocks confirms a read past the current watermark blocks
437
+
// until those bytes arrive, then returns them.
438
+
func TestPackCacheReadAheadBlocks(t *testing.T) {
439
+
ctx := context.Background()
440
+
const key = "r.git/objects/pack/ahead.pack"
441
+
data := bytes.Repeat([]byte("0123456789"), 1000) // 10 KiB
442
+
g := newGateReader(data)
443
+
stub := newGateStub()
444
+
stub.put(key, g)
445
+
pc := newTestPackCache(t, 0)
446
+
447
+
f, err := pc.open(ctx, stub, "bucket", key, "n")
448
+
if err != nil {
449
+
t.Fatalf("open: %v", err)
450
+
}
451
+
defer f.Close()
452
+
453
+
g.release(100)
454
+
// Read at offset 5000, far past the released watermark: must stay pending.
455
+
ch := readAtChan(f, 64, 5000)
456
+
select {
457
+
case res := <-ch:
458
+
t.Fatalf("ReadAt past watermark returned early: n=%d err=%v", res.n, res.err)
459
+
case <-time.After(200 * time.Millisecond):
460
+
// good, still blocked
461
+
}
462
+
463
+
// Release enough to cover the requested range; the read now completes.
464
+
g.releaseAll()
465
+
select {
466
+
case res := <-ch:
467
+
if res.err != nil && res.err != io.EOF {
468
+
t.Fatalf("ReadAt after release: %v", res.err)
469
+
}
470
+
if !bytes.Equal(res.p, data[5000:5064]) {
471
+
t.Fatalf("ReadAt bytes mismatch")
472
+
}
473
+
case <-time.After(2 * time.Second):
474
+
t.Fatal("ReadAt stayed blocked after data arrived")
475
+
}
476
+
}
477
+
478
+
// TestPackCacheBelowWindowReadsDisk confirms that once the RAM window has
479
+
// scrolled past an offset, reads of that offset are served from disk correctly.
480
+
func TestPackCacheBelowWindowReadsDisk(t *testing.T) {
481
+
ctx := context.Background()
482
+
const key = "r.git/objects/pack/big.pack"
483
+
// Larger than ringCap so the front of the object is dropped from RAM.
484
+
data := bytes.Repeat([]byte("Z"), ringCap+(1<<20))
485
+
for i := range data { // make bytes position-dependent to catch offset bugs
486
+
data[i] = byte(i % 251)
487
+
}
488
+
stub := newPackStub(map[string][]byte{key: data})
489
+
pc := newTestPackCache(t, 0)
490
+
491
+
f, err := pc.open(ctx, stub, "bucket", key, "n")
492
+
if err != nil {
493
+
t.Fatalf("open: %v", err)
494
+
}
495
+
defer f.Close()
496
+
497
+
// Force the whole download (synchronous body), so the window is fully
498
+
// trimmed and the entry is done; offset 0 must come from disk.
499
+
if got := mustReadAll(t, f); !bytes.Equal(got, data) {
500
+
t.Fatalf("full read mismatch")
501
+
}
502
+
p := make([]byte, 128)
503
+
if _, err := f.ReadAt(p, 0); err != nil && err != io.EOF {
504
+
t.Fatalf("ReadAt offset 0 from disk: %v", err)
505
+
}
506
+
if !bytes.Equal(p, data[:128]) {
507
+
t.Fatalf("disk-served bytes mismatch at offset 0")
508
+
}
509
+
// A mid-object offset below the (now empty) window, too.
510
+
off := int64(ringCap / 2)
511
+
if _, err := f.ReadAt(p, off); err != nil && err != io.EOF {
512
+
t.Fatalf("ReadAt mid offset from disk: %v", err)
513
+
}
514
+
if !bytes.Equal(p, data[off:off+128]) {
515
+
t.Fatalf("disk-served bytes mismatch at offset %d", off)
516
+
}
517
+
}
518
+
519
+
// TestPackCacheStreamError confirms a mid-stream download error surfaces to
520
+
// readers and the entry is dropped so a later open re-downloads.
521
+
func TestPackCacheStreamError(t *testing.T) {
522
+
ctx := context.Background()
523
+
const key = "r.git/objects/pack/err.pack"
524
+
data := bytes.Repeat([]byte("E"), 4096)
525
+
g := newGateReader(data)
526
+
boom := errors.New("boom")
527
+
g.errAt = 1024
528
+
g.failErr = boom
529
+
stub := newGateStub()
530
+
stub.put(key, g)
531
+
pc := newTestPackCache(t, 0)
532
+
533
+
f, err := pc.open(ctx, stub, "bucket", key, "n")
534
+
if err != nil {
535
+
t.Fatalf("open: %v", err)
536
+
}
537
+
538
+
// Release past the error point; the pump reads 1024 bytes then errors.
539
+
g.release(4096)
540
+
541
+
// A read past the failed watermark eventually returns the error.
542
+
deadline := time.Now().Add(2 * time.Second)
543
+
var readErr error
544
+
for time.Now().Before(deadline) {
545
+
p := make([]byte, 64)
546
+
_, readErr = f.ReadAt(p, 2048)
547
+
if readErr != nil {
548
+
break
549
+
}
550
+
time.Sleep(5 * time.Millisecond)
551
+
}
552
+
if !errors.Is(readErr, boom) {
553
+
t.Fatalf("read after stream error: err = %v, want %v", readErr, boom)
554
+
}
555
+
f.Close()
556
+
557
+
// The failed entry was dropped: a later open re-fetches. Use a fresh,
558
+
// fully-releasable gate this time.
559
+
g2 := newGateReader(data)
560
+
g2.releaseAll()
561
+
stub.put(key, g2)
562
+
f2, err := pc.open(ctx, stub, "bucket", key, "n2")
563
+
if err != nil {
564
+
t.Fatalf("reopen after error: %v", err)
565
+
}
566
+
if got := mustReadAll(t, f2); !bytes.Equal(got, data) {
567
+
t.Fatalf("reopen bytes mismatch")
568
+
}
569
+
f2.Close()
570
+
if n := stub.getCount(key); n != 2 {
571
+
t.Errorf("GetObject count = %d, want 2 (re-fetched after stream error)", n)
572
+
}
573
+
}
574
+
575
+
// TestPackCacheConcurrentStreamingReaders confirms many readers opened while a
576
+
// pack is still downloading all observe the full, correct bytes, with a single
577
+
// GetObject.
578
+
func TestPackCacheConcurrentStreamingReaders(t *testing.T) {
579
+
ctx := context.Background()
580
+
const key = "r.git/objects/pack/concurrent.pack"
581
+
data := bytes.Repeat([]byte("CONCURRENT"), 2000) // 20 KiB
582
+
g := newGateReader(data)
583
+
stub := newGateStub()
584
+
stub.put(key, g)
585
+
pc := newTestPackCache(t, 0)
586
+
587
+
const readers = 16
588
+
files := make([]billy.File, readers)
589
+
for i := range files {
590
+
f, err := pc.open(ctx, stub, "bucket", key, "n")
591
+
if err != nil {
592
+
t.Fatalf("open %d: %v", i, err)
593
+
}
594
+
files[i] = f
595
+
}
596
+
597
+
// Drip the body out in chunks while readers race to read it.
598
+
go func() {
599
+
for released := 0; released < len(data); released += 1024 {
600
+
g.release(1024)
601
+
time.Sleep(time.Millisecond)
602
+
}
603
+
g.releaseAll()
604
+
}()
605
+
606
+
var wg sync.WaitGroup
607
+
errs := make([]error, readers)
608
+
for i := range files {
609
+
wg.Add(1)
610
+
go func(i int) {
611
+
defer wg.Done()
612
+
got := make([]byte, len(data))
613
+
if _, err := io.ReadFull(io.NewSectionReader(asReaderAt(files[i]), 0, int64(len(data))), got); err != nil {
614
+
errs[i] = err
615
+
return
616
+
}
617
+
if !bytes.Equal(got, data) {
618
+
errs[i] = errors.New("bytes mismatch")
619
+
}
620
+
}(i)
621
+
}
622
+
wg.Wait()
623
+
624
+
for i, err := range errs {
625
+
if err != nil {
626
+
t.Errorf("reader %d: %v", i, err)
627
+
}
628
+
files[i].Close()
629
+
}
630
+
if n := stub.getCount(key); n != 1 {
631
+
t.Errorf("GetObject count = %d, want 1", n)
632
+
}
633
+
}
634
+
635
+
// TestPackCacheClosedDuringStreaming confirms a handle closed mid-download
636
+
// returns os.ErrClosed and does not wedge the pump.
637
+
func TestPackCacheClosedDuringStreaming(t *testing.T) {
638
+
ctx := context.Background()
639
+
const key = "r.git/objects/pack/closing.pack"
640
+
data := bytes.Repeat([]byte("X"), 8192)
641
+
g := newGateReader(data)
642
+
stub := newGateStub()
643
+
stub.put(key, g)
644
+
pc := newTestPackCache(t, 0)
645
+
646
+
f, err := pc.open(ctx, stub, "bucket", key, "n")
647
+
if err != nil {
648
+
t.Fatalf("open: %v", err)
649
+
}
650
+
g.release(100)
651
+
if err := f.Close(); err != nil {
652
+
t.Fatalf("close: %v", err)
653
+
}
654
+
if _, err := f.ReadAt(make([]byte, 1), 0); !errors.Is(err, os.ErrClosed) {
655
+
t.Errorf("ReadAt after close: err = %v, want os.ErrClosed", err)
656
+
}
657
+
g.releaseAll()
658
+
}
659
+
264
660
func TestIsPackCacheable(t *testing.T) {
265
661
tests := map[string]bool{
266
662
"r.git/objects/pack/pack-1.pack": true,
···
276
672
}
277
673
}
278
674
}
675
+
676
+
// asReaderAt adapts a billy.File to io.ReaderAt for use with io.SectionReader.
677
+
func asReaderAt(f billy.File) io.ReaderAt { return f }