Git backed by object storage because you can't stop me
git
object-storage
kefka
feat(s3fs): implement lazy, range-based reads with read-ahead window
Replace eager GetObject+ReadAll buffering with lazy on-demand fetching.
Files now stream their bodies only when read, and ReadAt uses S3 Range
requests backed by a 1 MiB read-ahead window. This prevents whole-object
buffering in RAM (critical for large pack files since feat: keep pushed
packs whole) and reduces redundant HeadObject calls by deriving metadata
from the GetObject response.
Sequential Read opens a body at Open or on first read (depending on cache
hit); body streams on demand and reopens after Seek. ReadAt issues Range
requests; nearby offsets are served from the window without new S3 calls.
The redundant HeadObject is dropped from the main path; only Stat on a
never-read handle falls back to it (effectively unreachable in normal flow).
The public billy.File API is unchanged; this is a performance and memory
optimization. All three transport protocols (HTTP, git://, SSH) exercise
real pack/idx reads through go-git's FSObject and Scanner; the full test
suite passes.
See docs/plans/s3fs-lazy-read.md for the design.
Assisted-by: Claude Haiku 4.5 via claude.ai
Signed-off-by: Xe Iaso <me@xeiaso.net>
+820
-62
+202
docs/plans/s3fs-lazy-read.md
+202
docs/plans/s3fs-lazy-read.md
···
1
+
# Lazy, range-based reads for s3fs
2
+
3
+
## Context
4
+
5
+
Today `internal/s3fs/file.go`'s `newS3ReadFile` does a `GetObject` **and**
6
+
`ioutil.ReadAll`s the entire object body into a `*bytes.Reader` at `Open` time
7
+
(file.go:83-99). Every read handle therefore holds the whole object in RAM —
8
+
including whole pack files, which since `feat: keep pushed packs whole` can be
9
+
the entire repository. This wastes memory and forces a full download even when
10
+
the consumer reads only a few bytes (e.g. a 1-byte `FSObject` probe, or an
11
+
`.idx` fanout lookup).
12
+
13
+
The goal: opening a file in read mode should **not** download anything. A
14
+
sequential `Read` should lazily start a `GetObject` whose body is read on
15
+
demand; `ReadAt` should fetch only the bytes it needs via S3 **Range** requests,
16
+
backed by a bounded in-memory read-ahead window so the many small sequential
17
+
`ReadAt` calls go-git issues (zlib ~512B reads, idx binary-search lookups) don't
18
+
amplify into hundreds of tiny GETs.
19
+
20
+
### Access patterns this must serve (from go-git v6)
21
+
22
+
- **Loose objects**: sequential `Read` start→finish → streaming body is ideal.
23
+
- **Pack `.pack`**: `FSObject.Reader` probes with a 1-byte `ReadAt` and reopens
24
+
the file on `os.ErrClosed`, then `io.NewSectionReader(...).ReadAt` at the
25
+
object offset; the `Scanner` uses `Seek`+sequential `Read`. → both paths.
26
+
- **Index `.idx`**: pure `ReadAt` at arbitrary offsets (header, fanout, hash and
27
+
offset tables). → random access, bursty-sequential.
28
+
29
+
## Approach
30
+
31
+
Rewrite `s3ReadFile` to be lazy with two independent, access-pattern-matched
32
+
mechanisms, and **drop both** the eager `GetObject`/`ReadAll` _and_ the dedicated
33
+
`HeadObject` from the read path.
34
+
35
+
### Metadata without a dedicated HeadObject
36
+
37
+
`enrichedFileInfo` only reads `ContentLength`, `LastModified`, and `Metadata`
38
+
(fileinfo.go:50-96) — all of which `s3.GetObjectOutput` also carries. So instead
39
+
of a standalone `HeadObject`, derive the metadata from whatever fetch happens
40
+
first:
41
+
42
+
- If `basic.go` supplied a cache-resident `head` (the `headInfo` path), use it
43
+
as today — zero fetches.
44
+
- Otherwise leave `head == nil` / `size == -1` (unknown) and fetch **nothing**
45
+
at open. The first `Read`/`ReadAt` does a `GetObject` anyway; build a synthetic
46
+
`*s3.HeadObjectOutput` from its response (`ContentLength`, `LastModified`,
47
+
`Metadata`, `ETag`, `ContentType`) and populate `head`/`size`.
48
+
- For a **non-ranged** GET (sequential `Read` from offset 0), `ContentLength`
49
+
_is_ the full size.
50
+
- For a **ranged** GET (`ReadAt`, or `Read` after a `Seek`), `ContentLength`
51
+
is the range length; parse the full size from `GetObjectOutput.ContentRange`
52
+
(`bytes <start>-<end>/<total>` → `total`). Add a tiny `parseContentRange`
53
+
helper.
54
+
- `Stat()` falls back to a lazy `HeadObject` **only** when `head` is still nil
55
+
(a consumer stats without ever reading and the cache didn't seed it). This is
56
+
the sole residual `HeadObject`, and it's rare.
57
+
58
+
Net effect on the uncached open→read path: **one** `GetObject` total, down from
59
+
today's `HeadObject` + full `GetObject`.
60
+
61
+
### New `s3ReadFile` shape (`internal/s3fs/file.go`)
62
+
63
+
Replace the `reader *bytes.Reader` field. New fields (one `sync.Mutex` guards
64
+
all mutable state for simplicity; reads on a single handle are effectively
65
+
serial):
66
+
67
+
```
68
+
client s3Client
69
+
bucket, key, name string
70
+
head *s3.HeadObjectOutput // nil until first fetch (or cache-supplied)
71
+
size int64 // full object size; -1 = unknown until first fetch
72
+
mu sync.Mutex
73
+
closed bool
74
+
75
+
// sequential streaming path (Read/Seek)
76
+
pos int64 // logical cursor
77
+
body io.ReadCloser // open GetObject body, nil until first Read
78
+
bodyPos int64 // object offset the body currently sits at
79
+
80
+
// random-access read-ahead window (ReadAt)
81
+
win []byte // cached chunk
82
+
winStart int64 // object offset of win[0]
83
+
```
84
+
85
+
`newS3ReadFile`: if `head` is supplied, set `size = *head.ContentLength`;
86
+
otherwise set `head = nil`, `size = -1`. Either way return immediately — **no
87
+
GetObject, no HeadObject**.
88
+
89
+
A shared `adoptMeta(out, ranged)` helper populates `head`/`size` from a
90
+
`*s3.GetObjectOutput` the first time one comes back (no-op if `head` already
91
+
set): synthesize `s3.HeadObjectOutput{ContentLength, LastModified, Metadata, …}`;
92
+
set `size` from `ContentLength` when not ranged, else from
93
+
`parseContentRange(out.ContentRange)`.
94
+
95
+
### `Read(p)` — lazy streaming body
96
+
97
+
- Return `os.ErrClosed` if closed.
98
+
- If `size >= 0` and `pos >= size`, return `0, io.EOF`.
99
+
- If `body == nil`: issue `GetObject` with `Range: bytes=<pos>-` (omit `Range`
100
+
when `pos == 0`), `observeS3("GetObject", …)`, `adoptMeta` from the response,
101
+
set `body`, `bodyPos = pos`.
102
+
- Read from `body` into `p`, advance `pos` and `bodyPos` by `n`. On the body's
103
+
`io.EOF`, return it through.
104
+
105
+
### `Seek(offset, whence)` — drop body on move
106
+
107
+
- Return `os.ErrClosed` if closed.
108
+
- Compute new absolute position (`SeekStart`/`SeekCurrent`/`SeekEnd` using
109
+
`size`). Reject negative.
110
+
- If `body != nil` and the new position `!= pos`, **close `body` and nil it**
111
+
(next `Read` reopens with the new `Range`). Set `pos`.
112
+
113
+
### `ReadAt(p, off)` — Range request + read-ahead window
114
+
115
+
- Return `os.ErrClosed` if closed (preserves the `FSObject` reopen contract).
116
+
- If `size >= 0` and `off >= size`, return `0, io.EOF`.
117
+
- Serve from `win` when `[off, off+len(p))` is covered by
118
+
`[winStart, winStart+len(win))`: `copy` and return.
119
+
- On miss: fetch `chunk := max(len(p), readChunkSize)` bytes (clamped to `size`
120
+
when known) starting at `off` via `GetObject` with
121
+
`Range: bytes=<off>-<off+chunk-1>`, `observeS3("GetObject", …)`, `adoptMeta`
122
+
from the response (learns `size` from `ContentRange` on the first call),
123
+
`io.ReadFull` the body into a new `win`, set `winStart = off`, then `copy`
124
+
into `p`.
125
+
- EOF handling **without pre-known size**: if S3 returns `416 InvalidRange`
126
+
(`off` at/after EOF) treat as `0, io.EOF` — and the `416` response still
127
+
carries the total via `ContentRange`, so `adoptMeta` can set `size`. If the
128
+
satisfied length `< len(p)` because the range hit EOF, return the bytes copied
129
+
plus `io.EOF` (honour the `io.ReaderAt` contract: short read ⇒ non-nil error).
130
+
131
+
`readChunkSize` is a package const defaulting to `1 << 20` (1 MiB). RAM per open
132
+
handle is bounded to one chunk (+ at most one streaming body buffer), not the
133
+
whole object. (No flag/Option for now — confirmed not needed.)
134
+
135
+
### `Stat()` — lazy head fallback
136
+
137
+
- If `head != nil` (cache-seeded or already adopted from a fetch), return
138
+
`enrichedFileInfo{HeadObjectOutput: *head, …}` as today.
139
+
- If `head == nil` (nothing read yet, no cache seed), do a one-time lazy
140
+
`HeadObject`, store it, and return it. This is the only surviving
141
+
`HeadObject` call.
142
+
143
+
### `Close()`
144
+
145
+
- Idempotent: `ErrFileClosed` if already closed.
146
+
- Close `body` if non-nil, drop `win`, set `closed = true`. `Read`/`ReadAt`/
147
+
`Seek` then return `os.ErrClosed` — exactly the contract `TestS3ReadFileClosed`
148
+
and `FSObject` depend on.
149
+
150
+
### Small helpers
151
+
152
+
- Range header formatter: `bytes=start-` and `bytes=start-end`. AWS SDK
153
+
`s3.GetObjectInput.Range` is a `*string`.
154
+
- `parseContentRange(s *string) int64`: parse the `/<total>` suffix of a
155
+
`bytes start-end/total` header; returns -1 if absent/unparseable.
156
+
- `adoptMeta(out *s3.GetObjectOutput, ranged bool)`: build and store the
157
+
synthetic `head` + `size` if not already set.
158
+
159
+
## Files to modify
160
+
161
+
- **`internal/s3fs/file.go`** — the rewrite above (`s3ReadFile` struct, `newS3ReadFile`,
162
+
`Read`, `ReadAt`, `Seek`, `Close`; `Stat`/`Name`/`Write*`/`Lock` unchanged).
163
+
`s3WriteFile`, `s3MultipartUploadFile`, `s3DirFile` untouched.
164
+
- **`internal/s3fs/file_test.go`** — `TestS3ReadFileClosed` builds
165
+
`&s3ReadFile{name: "k", reader: bytes.NewReader(...)}`; update the literal to
166
+
the new struct (e.g. set `closed` via `Close()` on a handle with a stub
167
+
client / non-nil `size`) while keeping the three `os.ErrClosed` subtests.
168
+
169
+
## Things that stay the same
170
+
171
+
- The `tempReadFile` / `tempBuffer` path (tempfs.go) is a separate read path for
172
+
in-flight temp packs — unchanged.
173
+
- The `OpenFile` resolution logic in `basic.go` (cache `resolve`, `headInfo`,
174
+
directory probe) — unchanged; it still hands `head` (or nil) to
175
+
`newS3ReadFile`.
176
+
- `ListingCache`, metrics observer wiring, `s3Client` interface — unchanged
177
+
(`GetObjectInput.Range` already exists on the SDK type).
178
+
179
+
## Verification
180
+
181
+
1. `go build ./...` and `go test ./internal/s3fs/...` — unit tests, including
182
+
the updated `TestS3ReadFileClosed`.
183
+
2. Add a table-driven test (follow the **`xe-go:go-table-driven-tests`** skill
184
+
for structure/naming) with the existing `stubClient` (listingcache_test.go)
185
+
extended to honour the `Range` field (return the sliced bytes and a matching
186
+
`ContentRange`/`ContentLength`, and `416` for out-of-range), asserting:
187
+
- open issues **no** `HeadObject` and **no** `GetObject`;
188
+
- `Read` issues the first `GetObject` lazily, then streams; `Stat` after it
189
+
reports the size derived from the response (no `HeadObject`);
190
+
- `ReadAt` learns `size` from `ContentRange` and serves a second nearby
191
+
offset from the window without a second `GetObject`;
192
+
- `ReadAt` past EOF returns `io.EOF` (via short read / `416`);
193
+
- `Stat` on a never-read handle triggers exactly one lazy `HeadObject`;
194
+
- post-`Close` calls return `os.ErrClosed`.
195
+
3. `go test ./cmd/objgitd/...` — full protocol tests (clone/push over HTTP,
196
+
git://, SSH) exercise real pack/idx reads through go-git's `FSObject` and
197
+
`Scanner`, validating the streaming + range paths end to end (requires `git`
198
+
on PATH).
199
+
4. Manual smoke (optional): run `./objgitd -bucket $BUCKET -allow-push`, push a
200
+
repo, clone it back, and confirm pack/idx reads succeed while watching
201
+
`objgit_s3_*` metrics for a sane GetObject count (read-ahead should keep it
202
+
far below one-per-ReadAt).
+302
-61
internal/s3fs/file.go
+302
-61
internal/s3fs/file.go
···
5
5
"context"
6
6
"errors"
7
7
"fmt"
8
+
"io"
8
9
"io/fs"
9
-
"io/ioutil"
10
10
"os"
11
+
"strconv"
11
12
"strings"
13
+
"sync"
12
14
"syscall"
13
15
"time"
14
16
17
+
"github.com/aws/aws-sdk-go-v2/aws"
15
18
"github.com/aws/aws-sdk-go-v2/service/s3"
19
+
"github.com/aws/smithy-go"
16
20
"go.uber.org/atomic"
17
21
"tangled.org/xeiaso.net/objgit/internal/s3fs/unixmeta"
18
22
)
···
45
49
ErrCantReadFromWriteOnly = errors.New("can't read from write-only file")
46
50
)
47
51
48
-
// s3ReadFile implements billy.File for S3, and represents a file opened in read mode.
52
+
// readChunkSize is the amount ReadAt fetches per Range request on a window
53
+
// miss. The fetched chunk is cached so the many small, sequential ReadAt calls
54
+
// go-git issues (zlib reads ~512B at a time; idx fanout/hash lookups) coalesce
55
+
// into a handful of GETs instead of one request per call. RAM per open handle
56
+
// is bounded to one chunk (plus at most one in-flight streaming body), not the
57
+
// whole object.
58
+
const readChunkSize = 1 << 20 // 1 MiB
59
+
60
+
// s3ReadFile implements billy.File for S3, representing a file opened in read
61
+
// mode. It fetches lazily: no object bytes are buffered at Open. Sequential
62
+
// Read streams a GetObject body read on demand; ReadAt issues Range requests
63
+
// backed by a small read-ahead window. See docs/plans and CLAUDE.md.
49
64
//
50
-
// Upon creation, the file is loaded from S3.
65
+
// All mutable state is guarded by mu. The io.ReaderAt contract (ReadAt is
66
+
// independent of the read cursor and safe for concurrent use) is honoured:
67
+
// ReadAt never touches pos/body, and a ReadAt drops any open sequential body so
68
+
// a later Read reopens it at pos.
51
69
type s3ReadFile struct {
52
-
client s3Client // S3 SDK client
53
-
bucket string // S3 bucket name
54
-
key string // File object's key in S3
55
-
name string // Root-relative path as presented to Open
70
+
client s3Client // S3 SDK client
71
+
bucket string // S3 bucket name
72
+
key string // File object's key in S3
73
+
name string // Root-relative path as presented to Open
74
+
75
+
mu sync.Mutex
56
76
closed bool // Is the file closed?
57
-
reader *bytes.Reader // Buffer for file contents
58
-
head *s3.HeadObjectOutput // File metadata from S3
77
+
head *s3.HeadObjectOutput // File metadata; nil until first fetch (or cache-supplied)
78
+
size int64 // Full object size; -1 until known
79
+
80
+
// Sequential streaming path (Read/Seek).
81
+
pos int64 // Logical read cursor.
82
+
body io.ReadCloser // Open GetObject body, nil until first Read (or set at open on the uncached path).
83
+
bodyPos int64 // Object offset the body currently sits at.
84
+
85
+
// Random-access read-ahead window (ReadAt).
86
+
win []byte // Cached chunk, nil until the first ReadAt miss.
87
+
winStart int64 // Object offset of win[0].
59
88
}
60
89
61
90
// newS3ReadFile creates a new s3ReadFile. key is the full S3 object key; name
62
-
// is the root-relative path the caller passed to Open (returned by Name). When
63
-
// head is non-nil it is used as-is — the caller already has the object's
64
-
// metadata (e.g. from the listing-cache head precache) so the redundant
65
-
// HeadObject round-trip is skipped; pass nil to fetch it here.
91
+
// is the root-relative path the caller passed to Open (returned by Name).
92
+
//
93
+
// When head is non-nil the caller already resolved the object's existence and
94
+
// metadata (e.g. from the listing cache), so no I/O happens here — the body is
95
+
// fetched lazily on the first Read. When head is nil a GetObject is issued now:
96
+
// it both confirms existence (callers rely on a NoSuchKey error to fall back to
97
+
// a directory probe) and supplies the metadata via its response headers, so the
98
+
// redundant HeadObject round-trip is gone. The body is kept for streaming, not
99
+
// read into memory.
66
100
func newS3ReadFile(client s3Client, bucket, key, name string, head *s3.HeadObjectOutput) (*s3ReadFile, error) {
67
-
// Create the context
68
-
ctx := context.TODO() // TODO: How can user-supplied contexts be supported?
101
+
f := &s3ReadFile{
102
+
client: client,
103
+
bucket: bucket,
104
+
key: key,
105
+
name: name,
106
+
size: -1,
107
+
}
69
108
70
-
if head == nil {
71
-
start := time.Now()
72
-
ho, err := client.HeadObject(ctx, &s3.HeadObjectInput{
73
-
Bucket: new(bucket),
74
-
Key: new(key),
75
-
})
76
-
observeS3("HeadObject", start, err)
77
-
if err != nil {
78
-
return nil, &os.PathError{Op: "read", Path: key, Err: err}
79
-
}
80
-
head = ho
109
+
if head != nil {
110
+
f.head = head
111
+
f.size = aws.ToInt64(head.ContentLength)
112
+
return f, nil
81
113
}
82
114
83
-
// Run the GetObject operation
115
+
// No cached metadata: open the object now. This GetObject is the existence
116
+
// check (its error drives the caller's directory probe) and the metadata
117
+
// source; the body streams lazily.
84
118
start := time.Now()
85
-
res, err := client.GetObject(ctx, &s3.GetObjectInput{
119
+
res, err := client.GetObject(context.TODO(), &s3.GetObjectInput{
86
120
Bucket: &bucket,
87
121
Key: &key,
88
122
})
···
90
124
if err != nil {
91
125
return nil, fmt.Errorf("unable to perform GetObject operation: %w", err)
92
126
}
127
+
f.adoptMeta(res, false)
128
+
f.body = res.Body
129
+
f.bodyPos = 0
130
+
return f, nil
131
+
}
93
132
94
-
// Read the file contents and store in a bytes reader
95
-
buf, err := ioutil.ReadAll(res.Body)
133
+
// adoptMeta records the object's full size and a synthetic HeadObjectOutput from
134
+
// a GetObject response the first time one is seen. ranged reports whether the
135
+
// response answered a Range request: a ranged response's ContentLength is the
136
+
// range length, so the full size comes from its Content-Range header instead. If
137
+
// the full size cannot be determined (an unparseable ranged response), head is
138
+
// left nil so Stat falls back to a HeadObject.
139
+
func (f *s3ReadFile) adoptMeta(out *s3.GetObjectOutput, ranged bool) {
140
+
if f.head != nil {
141
+
return
142
+
}
143
+
size := aws.ToInt64(out.ContentLength)
144
+
if ranged {
145
+
size = parseContentRange(out.ContentRange)
146
+
}
147
+
if size < 0 {
148
+
return
149
+
}
150
+
f.size = size
151
+
lastMod := out.LastModified
152
+
if lastMod == nil {
153
+
lastMod = aws.Time(time.Time{})
154
+
}
155
+
f.head = &s3.HeadObjectOutput{
156
+
ContentLength: aws.Int64(size),
157
+
LastModified: lastMod,
158
+
Metadata: out.Metadata,
159
+
ETag: out.ETag,
160
+
ContentType: out.ContentType,
161
+
}
162
+
}
163
+
164
+
// parseContentRange extracts the total object size from a Content-Range header
165
+
// of the form "bytes <start>-<end>/<total>". It returns -1 when the header is
166
+
// absent, uses an unknown total ("*"), or cannot be parsed.
167
+
func parseContentRange(s *string) int64 {
168
+
if s == nil {
169
+
return -1
170
+
}
171
+
i := strings.LastIndex(*s, "/")
172
+
if i < 0 {
173
+
return -1
174
+
}
175
+
total, err := strconv.ParseInt((*s)[i+1:], 10, 64)
96
176
if err != nil {
97
-
return nil, fmt.Errorf("unable to read file body: %w", err)
177
+
return -1
98
178
}
99
-
reader := bytes.NewReader(buf)
179
+
return total
180
+
}
100
181
101
-
// Return the file
102
-
return &s3ReadFile{
103
-
client: client,
104
-
bucket: bucket,
105
-
key: key,
106
-
name: name,
107
-
reader: reader,
108
-
head: head,
109
-
}, nil
182
+
// rangeHeader formats an HTTP Range header value. A negative end means
183
+
// open-ended ("bytes=start-").
184
+
func rangeHeader(start, end int64) string {
185
+
if end < 0 {
186
+
return fmt.Sprintf("bytes=%d-", start)
187
+
}
188
+
return fmt.Sprintf("bytes=%d-%d", start, end)
110
189
}
111
190
112
191
// Name returns the name of the file as presented to Open.
···
124
203
return 0, ErrCantWriteToReadOnly
125
204
}
126
205
127
-
// Read implements io.Reader for billy.File
206
+
// Read implements io.Reader for billy.File. The first Read (or the first after a
207
+
// Seek/ReadAt dropped the body) opens a GetObject body at the current position;
208
+
// subsequent reads stream from it.
128
209
func (f *s3ReadFile) Read(p []byte) (n int, err error) {
129
-
if f.closed || f.reader == nil {
210
+
f.mu.Lock()
211
+
defer f.mu.Unlock()
212
+
if f.closed {
130
213
return 0, os.ErrClosed
131
214
}
132
-
return f.reader.Read(p)
215
+
if f.size >= 0 && f.pos >= f.size {
216
+
return 0, io.EOF
217
+
}
218
+
if f.body == nil {
219
+
if err := f.openBodyLocked(f.pos); err != nil {
220
+
return 0, err
221
+
}
222
+
}
223
+
n, err = f.body.Read(p)
224
+
f.pos += int64(n)
225
+
f.bodyPos += int64(n)
226
+
return n, err
227
+
}
228
+
229
+
// openBodyLocked starts a GetObject whose body begins at offset at. Callers hold
230
+
// f.mu.
231
+
func (f *s3ReadFile) openBodyLocked(at int64) error {
232
+
in := &s3.GetObjectInput{Bucket: &f.bucket, Key: &f.key}
233
+
if at > 0 {
234
+
in.Range = aws.String(rangeHeader(at, -1))
235
+
}
236
+
start := time.Now()
237
+
out, err := f.client.GetObject(context.TODO(), in)
238
+
observeS3("GetObject", start, err)
239
+
if err != nil {
240
+
return fmt.Errorf("unable to perform GetObject operation: %w", err)
241
+
}
242
+
f.adoptMeta(out, at > 0)
243
+
f.body = out.Body
244
+
f.bodyPos = at
245
+
return nil
133
246
}
134
247
135
-
// ReadAt implements io.ReaderAt for billy.File. It returns os.ErrClosed once the
136
-
// file is closed (rather than dereferencing the nil reader and panicking): go-git's
137
-
// packfile.FSObject.Reader probes with a 1-byte ReadAt and reopens the pack when it
138
-
// sees an os.ErrClosed-matching error, so a cache-resident object over a closed pack
139
-
// handle recovers instead of crashing.
248
+
// ReadAt implements io.ReaderAt for billy.File. It serves from a read-ahead
249
+
// window when possible and otherwise fetches a chunk via a Range request. It
250
+
// returns os.ErrClosed once the file is closed (rather than dereferencing a nil
251
+
// reader and panicking): go-git's packfile.FSObject.Reader probes with a 1-byte
252
+
// ReadAt and reopens the pack when it sees an os.ErrClosed-matching error, so a
253
+
// cache-resident object over a closed pack handle recovers instead of crashing.
140
254
func (f *s3ReadFile) ReadAt(p []byte, off int64) (n int, err error) {
141
-
if f.closed || f.reader == nil {
255
+
if len(p) == 0 {
256
+
return 0, nil
257
+
}
258
+
f.mu.Lock()
259
+
defer f.mu.Unlock()
260
+
if f.closed {
142
261
return 0, os.ErrClosed
143
262
}
144
-
return f.reader.ReadAt(p, off)
263
+
if off < 0 {
264
+
return 0, fmt.Errorf("s3fs: ReadAt negative offset %d", off)
265
+
}
266
+
267
+
// Random access abandons any sequential stream; Read reopens it lazily.
268
+
if f.body != nil {
269
+
f.body.Close()
270
+
f.body = nil
271
+
}
272
+
273
+
if f.size >= 0 && off >= f.size {
274
+
return 0, io.EOF
275
+
}
276
+
277
+
// Serve fully-covered requests straight from the window.
278
+
end := off + int64(len(p))
279
+
if f.win != nil && off >= f.winStart && end <= f.winStart+int64(len(f.win)) {
280
+
copy(p, f.win[off-f.winStart:end-f.winStart])
281
+
return len(p), nil
282
+
}
283
+
284
+
// Window miss: fetch a chunk (at least len(p)) starting at off.
285
+
chunk := max(int64(len(p)), readChunkSize)
286
+
last := off + chunk - 1
287
+
if f.size >= 0 && last > f.size-1 {
288
+
last = f.size - 1
289
+
}
290
+
291
+
in := &s3.GetObjectInput{
292
+
Bucket: &f.bucket,
293
+
Key: &f.key,
294
+
Range: aws.String(rangeHeader(off, last)),
295
+
}
296
+
start := time.Now()
297
+
out, gerr := f.client.GetObject(context.TODO(), in)
298
+
observeS3("GetObject", start, gerr)
299
+
if gerr != nil {
300
+
// A range starting at or past EOF is InvalidRange; treat as EOF.
301
+
if isInvalidRange(gerr) {
302
+
return 0, io.EOF
303
+
}
304
+
return 0, fmt.Errorf("unable to perform GetObject operation: %w", gerr)
305
+
}
306
+
f.adoptMeta(out, true)
307
+
buf, rerr := io.ReadAll(out.Body)
308
+
out.Body.Close()
309
+
if rerr != nil {
310
+
return 0, fmt.Errorf("unable to read range body: %w", rerr)
311
+
}
312
+
313
+
f.win = buf
314
+
f.winStart = off
315
+
n = copy(p, buf)
316
+
if n < len(p) {
317
+
// The range hit EOF before filling p.
318
+
return n, io.EOF
319
+
}
320
+
return n, nil
321
+
}
322
+
323
+
// isInvalidRange reports whether err is S3's 416 InvalidRange response, returned
324
+
// for a Range whose start is at or beyond the object's end.
325
+
func isInvalidRange(err error) bool {
326
+
var apiErr smithy.APIError
327
+
if errors.As(err, &apiErr) {
328
+
return apiErr.ErrorCode() == "InvalidRange"
329
+
}
330
+
return false
145
331
}
146
332
147
-
// Seek implements io.Seeker for billy.File
333
+
// Seek implements io.Seeker for billy.File. Seeking away from the body's current
334
+
// position drops it; the next Read reopens at the new offset.
148
335
func (f *s3ReadFile) Seek(offset int64, whence int) (int64, error) {
149
-
if f.closed || f.reader == nil {
336
+
f.mu.Lock()
337
+
defer f.mu.Unlock()
338
+
if f.closed {
150
339
return 0, os.ErrClosed
151
340
}
152
-
return f.reader.Seek(offset, whence)
341
+
342
+
var np int64
343
+
switch whence {
344
+
case io.SeekStart:
345
+
np = offset
346
+
case io.SeekCurrent:
347
+
np = f.pos + offset
348
+
case io.SeekEnd:
349
+
if f.size < 0 {
350
+
if err := f.headLocked(); err != nil {
351
+
return 0, err
352
+
}
353
+
}
354
+
np = f.size + offset
355
+
default:
356
+
return 0, fmt.Errorf("s3fs: Seek invalid whence %d", whence)
357
+
}
358
+
if np < 0 {
359
+
return 0, fmt.Errorf("s3fs: Seek negative position %d", np)
360
+
}
361
+
362
+
if f.body != nil && np != f.pos {
363
+
f.body.Close()
364
+
f.body = nil
365
+
}
366
+
f.pos = np
367
+
return np, nil
153
368
}
154
369
155
370
// Close implements io.Closer for billy.File
156
371
func (f *s3ReadFile) Close() error {
157
-
// Was the file already closed?
372
+
f.mu.Lock()
373
+
defer f.mu.Unlock()
158
374
if f.closed {
159
375
return ErrFileClosed
160
376
}
161
-
162
-
// Close the underlying file
163
-
f.reader = nil
164
-
165
-
// Mark the file as closed
377
+
if f.body != nil {
378
+
f.body.Close()
379
+
f.body = nil
380
+
}
381
+
f.win = nil
166
382
f.closed = true
167
-
168
383
return nil
169
384
}
170
385
···
185
400
}
186
401
187
402
func (f *s3ReadFile) Stat() (fs.FileInfo, error) {
403
+
f.mu.Lock()
404
+
defer f.mu.Unlock()
405
+
if f.head == nil {
406
+
if err := f.headLocked(); err != nil {
407
+
return nil, err
408
+
}
409
+
}
188
410
return enrichedFileInfo{
189
411
HeadObjectOutput: *f.head,
190
412
key: f.key,
191
413
mode: fs.ModePerm,
192
414
}, nil
415
+
}
416
+
417
+
// headLocked fetches object metadata via HeadObject and records it. It is the
418
+
// sole remaining HeadObject path, reached only when no fetch has populated head
419
+
// yet (e.g. Stat or SeekEnd before any read on a cache-supplied-less handle).
420
+
// Callers hold f.mu.
421
+
func (f *s3ReadFile) headLocked() error {
422
+
start := time.Now()
423
+
ho, err := f.client.HeadObject(context.TODO(), &s3.HeadObjectInput{
424
+
Bucket: &f.bucket,
425
+
Key: &f.key,
426
+
})
427
+
observeS3("HeadObject", start, err)
428
+
if err != nil {
429
+
return &os.PathError{Op: "stat", Path: f.key, Err: err}
430
+
}
431
+
f.head = ho
432
+
f.size = aws.ToInt64(ho.ContentLength)
433
+
return nil
193
434
}
194
435
195
436
// s3WriteFile stores a file opened in write mode and implements billy.File
+315
internal/s3fs/file_lazy_test.go
+315
internal/s3fs/file_lazy_test.go
···
1
+
package s3fs
2
+
3
+
import (
4
+
"bytes"
5
+
"context"
6
+
"errors"
7
+
"fmt"
8
+
"io"
9
+
"strconv"
10
+
"strings"
11
+
"sync/atomic"
12
+
"testing"
13
+
"time"
14
+
15
+
"github.com/aws/aws-sdk-go-v2/aws"
16
+
"github.com/aws/aws-sdk-go-v2/service/s3"
17
+
"github.com/aws/smithy-go"
18
+
)
19
+
20
+
// rangeClient is a content-aware s3Client stub for the lazy read path. Unlike
21
+
// stubClient (size-only), it stores real bytes and honours the Range header on
22
+
// GetObject: it slices the object, fills ContentRange/ContentLength, and returns
23
+
// a 416 InvalidRange error for a start at or past the object's end. It also
24
+
// reports, per object, how many body bytes the consumer has actually read, so
25
+
// tests can assert that Open does not drain the body.
26
+
type rangeClient struct {
27
+
objects map[string][]byte
28
+
29
+
gets atomic.Int64
30
+
heads atomic.Int64
31
+
read atomic.Int64 // body bytes the consumer has read across all GetObject responses
32
+
}
33
+
34
+
func newRangeClient(objects map[string][]byte) *rangeClient {
35
+
return &rangeClient{objects: objects}
36
+
}
37
+
38
+
// countingBody wraps a reader and tallies consumed bytes onto the client, so a
39
+
// test can prove the body is streamed lazily rather than buffered at Open.
40
+
type countingBody struct {
41
+
r io.Reader
42
+
c *rangeClient
43
+
}
44
+
45
+
func (b countingBody) Read(p []byte) (int, error) {
46
+
n, err := b.r.Read(p)
47
+
b.c.read.Add(int64(n))
48
+
return n, err
49
+
}
50
+
51
+
func (b countingBody) Close() error { return nil }
52
+
53
+
func (c *rangeClient) HeadObject(_ context.Context, in *s3.HeadObjectInput, _ ...func(*s3.Options)) (*s3.HeadObjectOutput, error) {
54
+
c.heads.Add(1)
55
+
data, ok := c.objects[aws.ToString(in.Key)]
56
+
if !ok {
57
+
return nil, notFound()
58
+
}
59
+
return &s3.HeadObjectOutput{
60
+
ContentLength: aws.Int64(int64(len(data))),
61
+
LastModified: aws.Time(time.Unix(0, 0)),
62
+
}, nil
63
+
}
64
+
65
+
func (c *rangeClient) GetObject(_ context.Context, in *s3.GetObjectInput, _ ...func(*s3.Options)) (*s3.GetObjectOutput, error) {
66
+
c.gets.Add(1)
67
+
data, ok := c.objects[aws.ToString(in.Key)]
68
+
if !ok {
69
+
return nil, notFound()
70
+
}
71
+
full := int64(len(data))
72
+
73
+
if in.Range == nil {
74
+
return &s3.GetObjectOutput{
75
+
Body: countingBody{r: bytes.NewReader(data), c: c},
76
+
ContentLength: aws.Int64(full),
77
+
LastModified: aws.Time(time.Unix(0, 0)),
78
+
}, nil
79
+
}
80
+
81
+
start, end, err := parseRange(aws.ToString(in.Range), full)
82
+
if err != nil {
83
+
return nil, err
84
+
}
85
+
if start >= full {
86
+
return nil, &smithy.GenericAPIError{Code: "InvalidRange", Message: "range start past end"}
87
+
}
88
+
if end > full-1 {
89
+
end = full - 1
90
+
}
91
+
slice := data[start : end+1]
92
+
return &s3.GetObjectOutput{
93
+
Body: countingBody{r: bytes.NewReader(slice), c: c},
94
+
ContentLength: aws.Int64(int64(len(slice))),
95
+
ContentRange: aws.String(fmt.Sprintf("bytes %d-%d/%d", start, end, full)),
96
+
LastModified: aws.Time(time.Unix(0, 0)),
97
+
}, nil
98
+
}
99
+
100
+
// parseRange parses "bytes=start-" and "bytes=start-end" headers.
101
+
func parseRange(h string, full int64) (start, end int64, err error) {
102
+
spec, ok := strings.CutPrefix(h, "bytes=")
103
+
if !ok {
104
+
return 0, 0, fmt.Errorf("bad range %q", h)
105
+
}
106
+
lo, hi, _ := strings.Cut(spec, "-")
107
+
start, err = strconv.ParseInt(lo, 10, 64)
108
+
if err != nil {
109
+
return 0, 0, err
110
+
}
111
+
if hi == "" {
112
+
return start, full - 1, nil
113
+
}
114
+
end, err = strconv.ParseInt(hi, 10, 64)
115
+
if err != nil {
116
+
return 0, 0, err
117
+
}
118
+
return start, end, nil
119
+
}
120
+
121
+
// the remaining s3Client methods are unused by these tests.
122
+
func (c *rangeClient) PutObject(context.Context, *s3.PutObjectInput, ...func(*s3.Options)) (*s3.PutObjectOutput, error) {
123
+
return &s3.PutObjectOutput{}, nil
124
+
}
125
+
func (c *rangeClient) ListObjectsV2(context.Context, *s3.ListObjectsV2Input, ...func(*s3.Options)) (*s3.ListObjectsV2Output, error) {
126
+
return &s3.ListObjectsV2Output{}, nil
127
+
}
128
+
func (c *rangeClient) DeleteObject(context.Context, *s3.DeleteObjectInput, ...func(*s3.Options)) (*s3.DeleteObjectOutput, error) {
129
+
return &s3.DeleteObjectOutput{}, nil
130
+
}
131
+
func (c *rangeClient) RenameObject(context.Context, *s3.CopyObjectInput, ...func(*s3.Options)) (*s3.CopyObjectOutput, error) {
132
+
return &s3.CopyObjectOutput{}, nil
133
+
}
134
+
func (c *rangeClient) CreateMultipartUpload(context.Context, *s3.CreateMultipartUploadInput, ...func(*s3.Options)) (*s3.CreateMultipartUploadOutput, error) {
135
+
return &s3.CreateMultipartUploadOutput{UploadId: aws.String("u")}, nil
136
+
}
137
+
func (c *rangeClient) UploadPart(context.Context, *s3.UploadPartInput, ...func(*s3.Options)) (*s3.UploadPartOutput, error) {
138
+
return &s3.UploadPartOutput{}, nil
139
+
}
140
+
func (c *rangeClient) CompleteMultipartUpload(context.Context, *s3.CompleteMultipartUploadInput, ...func(*s3.Options)) (*s3.CompleteMultipartUploadOutput, error) {
141
+
return &s3.CompleteMultipartUploadOutput{}, nil
142
+
}
143
+
144
+
// TestS3ReadFileLazyOpen covers the lazy/range read path: Open must not drain
145
+
// the body, the uncached path issues exactly one GetObject and no HeadObject,
146
+
// and the cache-supplied-head path issues no I/O until the first Read.
147
+
func TestS3ReadFileLazyOpen(t *testing.T) {
148
+
const key = "objects/pack/p.pack"
149
+
body := []byte("the quick brown fox jumps over the lazy dog")
150
+
151
+
tests := []struct {
152
+
name string
153
+
head *s3.HeadObjectOutput // nil = uncached path
154
+
wantOpenGet int64
155
+
}{
156
+
{
157
+
name: "uncached open issues one GetObject",
158
+
head: nil,
159
+
wantOpenGet: 1,
160
+
},
161
+
{
162
+
name: "cache-supplied head issues no IO at open",
163
+
head: &s3.HeadObjectOutput{ContentLength: aws.Int64(int64(len(body))), LastModified: aws.Time(time.Unix(0, 0))},
164
+
wantOpenGet: 0,
165
+
},
166
+
}
167
+
168
+
for _, tt := range tests {
169
+
t.Run(tt.name, func(t *testing.T) {
170
+
c := newRangeClient(map[string][]byte{key: body})
171
+
f, err := newS3ReadFile(c, "bucket", key, key, tt.head)
172
+
if err != nil {
173
+
t.Fatalf("newS3ReadFile: %v", err)
174
+
}
175
+
t.Cleanup(func() { f.Close() })
176
+
177
+
if got := c.gets.Load(); got != tt.wantOpenGet {
178
+
t.Fatalf("open GetObject = %d, want %d", got, tt.wantOpenGet)
179
+
}
180
+
if got := c.heads.Load(); got != 0 {
181
+
t.Fatalf("open HeadObject = %d, want 0", got)
182
+
}
183
+
if got := c.read.Load(); got != 0 {
184
+
t.Fatalf("open drained %d body bytes, want 0 (body must be lazy)", got)
185
+
}
186
+
187
+
// Reading streams the body and yields the full contents.
188
+
got, err := io.ReadAll(f)
189
+
if err != nil {
190
+
t.Fatalf("ReadAll: %v", err)
191
+
}
192
+
if !bytes.Equal(got, body) {
193
+
t.Fatalf("Read = %q, want %q", got, body)
194
+
}
195
+
196
+
// Stat reports the size derived from the response, no HeadObject.
197
+
fi, err := f.Stat()
198
+
if err != nil {
199
+
t.Fatalf("Stat: %v", err)
200
+
}
201
+
if fi.Size() != int64(len(body)) {
202
+
t.Fatalf("Stat size = %d, want %d", fi.Size(), len(body))
203
+
}
204
+
if got := c.heads.Load(); got != 0 {
205
+
t.Fatalf("HeadObject = %d after Read+Stat, want 0", got)
206
+
}
207
+
})
208
+
}
209
+
}
210
+
211
+
// TestS3ReadFileReadAt covers the random-access path: ReadAt learns the size
212
+
// from Content-Range, serves a nearby second offset from the read-ahead window
213
+
// without a second GetObject, and returns io.EOF past the end.
214
+
func TestS3ReadFileReadAt(t *testing.T) {
215
+
const key = "objects/pack/p.idx"
216
+
body := []byte("0123456789abcdefghijklmnopqrstuvwxyz")
217
+
c := newRangeClient(map[string][]byte{key: body})
218
+
219
+
// Cache-supplied head so Open does no I/O; ReadAt drives all GETs.
220
+
f, err := newS3ReadFile(c, "bucket", key, key,
221
+
&s3.HeadObjectOutput{ContentLength: aws.Int64(int64(len(body))), LastModified: aws.Time(time.Unix(0, 0))})
222
+
if err != nil {
223
+
t.Fatalf("newS3ReadFile: %v", err)
224
+
}
225
+
t.Cleanup(func() { f.Close() })
226
+
227
+
// First ReadAt fetches a chunk and serves from it.
228
+
buf := make([]byte, 4)
229
+
if n, err := f.ReadAt(buf, 2); err != nil || n != 4 {
230
+
t.Fatalf("ReadAt(_, 2) = (%d, %v), want (4, nil)", n, err)
231
+
}
232
+
if string(buf) != "2345" {
233
+
t.Fatalf("ReadAt(_, 2) = %q, want %q", buf, "2345")
234
+
}
235
+
gets := c.gets.Load()
236
+
if gets != 1 {
237
+
t.Fatalf("first ReadAt GetObject = %d, want 1", gets)
238
+
}
239
+
240
+
// A nearby second offset is covered by the window: no new GetObject.
241
+
if n, err := f.ReadAt(buf, 10); err != nil || n != 4 {
242
+
t.Fatalf("ReadAt(_, 10) = (%d, %v), want (4, nil)", n, err)
243
+
}
244
+
if string(buf) != "abcd" {
245
+
t.Fatalf("ReadAt(_, 10) = %q, want %q", buf, "abcd")
246
+
}
247
+
if c.gets.Load() != gets {
248
+
t.Fatalf("windowed ReadAt issued a GetObject: %d -> %d", gets, c.gets.Load())
249
+
}
250
+
251
+
// Reading past the end returns io.EOF.
252
+
if n, err := f.ReadAt(make([]byte, 4), int64(len(body))); !errors.Is(err, io.EOF) || n != 0 {
253
+
t.Fatalf("ReadAt past end = (%d, %v), want (0, io.EOF)", n, err)
254
+
}
255
+
}
256
+
257
+
// TestS3ReadFileReadAtUnknownSize exercises the head==nil-with-416 corner: when
258
+
// the size is not yet known and ReadAt starts past the end, the stub's 416
259
+
// InvalidRange must surface as io.EOF.
260
+
func TestS3ReadFileReadAtUnknownSize(t *testing.T) {
261
+
const key = "objects/small"
262
+
body := []byte("tiny")
263
+
c := newRangeClient(map[string][]byte{key: body})
264
+
265
+
// Uncached open issues a GetObject and adopts the (full) size, so force the
266
+
// unknown-size 416 path directly on a fresh handle.
267
+
f := &s3ReadFile{client: c, bucket: "bucket", key: key, name: key, size: -1}
268
+
t.Cleanup(func() { f.Close() })
269
+
270
+
if n, err := f.ReadAt(make([]byte, 2), 100); !errors.Is(err, io.EOF) || n != 0 {
271
+
t.Fatalf("ReadAt past end (unknown size) = (%d, %v), want (0, io.EOF)", n, err)
272
+
}
273
+
}
274
+
275
+
// TestS3ReadFileMissingKey confirms the uncached Open path still surfaces a
276
+
// NoSuchKey/NotFound error (basic.go relies on it for the directory probe).
277
+
func TestS3ReadFileMissingKey(t *testing.T) {
278
+
c := newRangeClient(map[string][]byte{})
279
+
if _, err := newS3ReadFile(c, "bucket", "absent", "absent", nil); err == nil {
280
+
t.Fatal("newS3ReadFile of an absent key: want error, got nil")
281
+
}
282
+
}
283
+
284
+
// TestS3ReadFileSeek covers Seek semantics: a Seek that moves the cursor drops
285
+
// the open body so the next Read reopens with a Range at the new position.
286
+
func TestS3ReadFileSeek(t *testing.T) {
287
+
const key = "blob"
288
+
body := []byte("HEADERthen the payload")
289
+
c := newRangeClient(map[string][]byte{key: body})
290
+
291
+
f, err := newS3ReadFile(c, "bucket", key, key, nil)
292
+
if err != nil {
293
+
t.Fatalf("newS3ReadFile: %v", err)
294
+
}
295
+
t.Cleanup(func() { f.Close() })
296
+
297
+
if _, err := f.Seek(6, io.SeekStart); err != nil {
298
+
t.Fatalf("Seek: %v", err)
299
+
}
300
+
got, err := io.ReadAll(f)
301
+
if err != nil {
302
+
t.Fatalf("ReadAll after seek: %v", err)
303
+
}
304
+
if want := body[6:]; !bytes.Equal(got, want) {
305
+
t.Fatalf("Read after Seek = %q, want %q", got, want)
306
+
}
307
+
308
+
// SeekEnd needs the size; for the uncached handle it was learned at open.
309
+
if pos, err := f.Seek(0, io.SeekEnd); err != nil || pos != int64(len(body)) {
310
+
t.Fatalf("Seek(0, SeekEnd) = (%d, %v), want (%d, nil)", pos, err, len(body))
311
+
}
312
+
if n, err := f.Read(make([]byte, 1)); !errors.Is(err, io.EOF) || n != 0 {
313
+
t.Fatalf("Read at EOF = (%d, %v), want (0, io.EOF)", n, err)
314
+
}
315
+
}
+1
-1
internal/s3fs/file_test.go
+1
-1
internal/s3fs/file_test.go
···
16
16
// (e.g. a post-receive hook reading the just-pushed commit).
17
17
func TestS3ReadFileClosed(t *testing.T) {
18
18
newClosed := func() *s3ReadFile {
19
-
f := &s3ReadFile{name: "k", reader: bytes.NewReader([]byte("hello"))}
19
+
f := &s3ReadFile{name: "k", size: 5}
20
20
if err := f.Close(); err != nil {
21
21
t.Fatalf("Close: %v", err)
22
22
}