Git backed by object storage because you can't stop me
git
object-storage
kefka
feat(s3fs): cache directory listings and object metadata via groupcache
Eliminate redundant HeadObject and ListObjectsV2 round-trips on repeated
Stat/Open operations against the same S3 prefixes. Implemented via two
groupcache groups with window-encoded TTL and per-prefix local generation
for precise in-process invalidation.
When a path's parent folder is first listed, the cache populates both:
- childEntry metadata (name, kind, size, mtime) for the entire folder
- Background prefetch of HeadObject for each file via detached goroutines
with semaphore bounding (maxPrefetchInFlight=64, configurable)
Stat/Open of a cached folder's absent child answers with zero round-trips
(negative hit from the listing). A present file skips HeadObject via the
prewarmed head cache and only GetObject-s the body. Positive Stat/ReadDir
results are never stale (immutable S3 content); cross-process and
post-restart staleness is bounded by the TTL window (-s3-cache-ttl, default
60s; <=0 disables caching entirely).
Local writes (Create, Remove, Rename, MkdirAll) bump the parent prefix's
generation on success, moving the cache key so the next read re-lists and
sees the change immediately (read-after-write correctness).
When -groupcache-self and -groupcache-peers are configured, listings are
shared across processes via consistent-hash peer pooling.
Changes:
- New: internal/s3fs/listingcache.go (ListingCache type, groupcache groups,
warmer goroutine for refresh at idle TTL boundaries)
- New: internal/s3fs/listingcache_test.go (comprehensive table-driven tests
with counting stub client, head prefetch tests, window TTL, warmer, chroot
integration)
- Modified: internal/s3fs/{filesystem,basic,file,dir,chroot}.go to wire cache
through Stat/Open/ReadDir/Remove/Rename/MkdirAll with invalidation on writes
- Modified: internal/metrics/metrics.go to export groupcache stats via
objgit_s3_listing_cache_* collectors
- Modified: cmd/objgitd/main.go to add -s3-cache-ttl, -s3-cache-refresh,
-s3-cache-idle, -s3-cache-size, -groupcache-self, -groupcache-peers,
-groupcache-bind flags; integrate cache into S3FS and errgroup
Assisted-by: Claude Opus 4.8 via Claude Code
Signed-off-by: Xe Iaso <me@xeiaso.net>
+1401
-58
+74
-1
cmd/objgitd/main.go
+74
-1
cmd/objgitd/main.go
···
12
12
"os"
13
13
"os/signal"
14
14
"runtime"
15
+
"strings"
15
16
"syscall"
16
17
"time"
17
18
···
40
41
41
42
allowHooks = flag.Bool("allow-hooks", false, "run .objgit/hooks/receive-pack in a sandbox after a successful push")
42
43
hookTimeout = flag.Duration("hook-timeout", 60*time.Second, "wall-clock limit for a single hook run")
44
+
45
+
s3CacheTTL = flag.Duration("s3-cache-ttl", 60*time.Second, "how long a cached S3 directory listing answers Stat/Open before a re-list; 0 disables the listing cache")
46
+
s3CacheRefresh = flag.Duration("s3-cache-refresh", 30*time.Second, "interval at which the listing-cache warmer re-fills hot prefixes; 0 disables the warmer")
47
+
s3CacheIdle = flag.Duration("s3-cache-idle", 10*time.Minute, "drop a prefix from the listing-cache warmer after this long without access")
48
+
s3CacheSize = flag.Int64("s3-cache-size", 64<<20, "groupcache LRU budget in bytes for the listing cache")
49
+
50
+
groupcacheSelf = flag.String("groupcache-self", "", "this node's groupcache base URL (e.g. http://10.0.0.1:8000); empty runs the cache single-process")
51
+
groupcachePeers = flag.String("groupcache-peers", "", "comma-separated groupcache peer base URLs (including self) for cross-process sharing")
52
+
groupcacheBind = flag.String("groupcache-bind", "", "TCP address to serve the groupcache peer endpoint; empty disables peer sharing")
43
53
)
44
54
45
55
func main() {
···
75
85
os.Exit(1)
76
86
}
77
87
78
-
fsys, err := s3fs.NewS3FS(client, *bucket)
88
+
var cache *s3fs.ListingCache
89
+
var fsOpts []s3fs.Option
90
+
if *s3CacheTTL > 0 {
91
+
var peers []string
92
+
if *groupcachePeers != "" {
93
+
peers = strings.Split(*groupcachePeers, ",")
94
+
}
95
+
cache = s3fs.NewListingCache(s3fs.CacheConfig{
96
+
TTL: *s3CacheTTL,
97
+
RefreshInterval: *s3CacheRefresh,
98
+
IdleTTL: *s3CacheIdle,
99
+
SizeBytes: *s3CacheSize,
100
+
Self: *groupcacheSelf,
101
+
Peers: peers,
102
+
}, client, *bucket, "/")
103
+
fsOpts = append(fsOpts, s3fs.WithListingCache(cache))
104
+
metrics.RegisterListingCache(func() metrics.ListingCacheStats {
105
+
s := cache.Stats()
106
+
return metrics.ListingCacheStats{
107
+
Gets: s.Gets, CacheHits: s.CacheHits, Loads: s.Loads,
108
+
LocalLoads: s.LocalLoads, PeerLoads: s.PeerLoads, LocalLoadErrs: s.LocalLoadErrs,
109
+
MainBytes: s.MainBytes, MainItems: s.MainItems, MainEvictions: s.MainEvictions,
110
+
HotBytes: s.HotBytes, HotItems: s.HotItems,
111
+
}
112
+
})
113
+
}
114
+
115
+
fsys, err := s3fs.NewS3FS(client, *bucket, fsOpts...)
79
116
if err != nil {
80
117
slog.Error("can't create s3fs", "bucket", *bucket, "err", err)
81
118
os.Exit(1)
···
97
134
"bucket", *bucket,
98
135
"allow_push", *allowPush,
99
136
"allow_hooks", *allowHooks,
137
+
"s3_cache_ttl", *s3CacheTTL,
138
+
"s3_cache_refresh", *s3CacheRefresh,
139
+
"groupcache_self", *groupcacheSelf,
100
140
)
101
141
102
142
g, gCtx := errgroup.WithContext(ctx)
143
+
144
+
if cache != nil {
145
+
g.Go(func() error {
146
+
cache.RunWarmer(gCtx)
147
+
return nil
148
+
})
149
+
150
+
if *groupcacheBind != "" {
151
+
handler := cache.PoolHandler()
152
+
if handler == nil {
153
+
slog.Error("-groupcache-bind set without -groupcache-self; peer sharing is disabled")
154
+
os.Exit(1)
155
+
}
156
+
ln, err := net.Listen("tcp", *groupcacheBind)
157
+
if err != nil {
158
+
slog.Error("can't listen", "groupcache_bind", *groupcacheBind, "err", err)
159
+
os.Exit(1)
160
+
}
161
+
srv := &http.Server{Handler: handler}
162
+
g.Go(func() error {
163
+
if err := srv.Serve(ln); err != nil && !errors.Is(err, http.ErrServerClosed) {
164
+
return err
165
+
}
166
+
return nil
167
+
})
168
+
g.Go(func() error {
169
+
<-gCtx.Done()
170
+
shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
171
+
defer cancel()
172
+
return srv.Shutdown(shutdownCtx)
173
+
})
174
+
}
175
+
}
103
176
104
177
if *metricsBind != "" {
105
178
ln, err := net.Listen("tcp", *metricsBind)
+267
docs/plans/s3fs-groupcache-listobjectsv2.md
+267
docs/plans/s3fs-groupcache-listobjectsv2.md
···
1
+
# Plan: groupcache-backed directory-listing cache for `internal/s3fs`
2
+
3
+
## Context
4
+
5
+
`objgitd` stores git repos as S3/Tigris objects. Every `Stat`/`Open` on a path that
6
+
**doesn't exist** costs up to **two** S3 round-trips: a `HeadObject` (→ `NoSuchKey`)
7
+
then a directory-probe `ListObjectsV2` (`internal/s3fs/basic.go:140`+`:163`; the probe
8
+
in `OpenFile` at `:98`). git does an enormous number of these — loose objects on a
9
+
packed repo, `packed-refs` vs. loose refs, `info/`, `config`, alternates, `.keep`/`.idx`
10
+
siblings — each paid in series against an object store.
11
+
12
+
The fix: cache **directory listings**, keyed by parent prefix, and answer `Stat`/`Open`
13
+
from them. A listing records every child name plus its kind/size/mtime, so a lookup of a
14
+
key whose parent prefix is cached answers with **zero** round-trips when the child is
15
+
absent (negative hit) or is a sub-directory, and with one `HeadObject`/`GetObject` (for
16
+
authoritative content/metadata, which listings don't carry) when it's a file. The first
17
+
`Stat`/`Open` touching an un-cached folder lists that **parent folder** in full and
18
+
populates the cache, warming every sibling at once.
19
+
20
+
**Backend (user decision): `github.com/golang/groupcache`,** so the listing cache is
21
+
shared across processes via groupcache's consistent-hash peer pool. groupcache is a
22
+
fill-once, **immutable** cache: no `Set`, no `Delete`, no TTL. We adapt to that with two
23
+
techniques baked into the cache **key**:
24
+
25
+
- **Window-encoded TTL.** The key carries a time window `floor(now/TTL)`. When the
26
+
window advances the key changes → groupcache miss → the getter re-lists from S3. Old
27
+
keys age out by LRU. Staleness is bounded by one TTL window.
28
+
- **Per-prefix local generation for precise in-process invalidation.** The key also
29
+
carries a process-local counter per prefix, bumped on every local write under that
30
+
prefix. A local write moves the key, so the next read in this process re-lists and
31
+
sees its own write immediately — recovering read-after-write correctness that
32
+
groupcache alone cannot give (no delete). The stale entry under the old key is simply
33
+
never queried again by us.
34
+
35
+
**Accepted limitation (explicit user choice):** cross-process and post-restart staleness
36
+
is bounded by the TTL window — another process (or this one after a restart that resets
37
+
the local generation) may read a just-written object as "not found" for up to one
38
+
window. This is safe for git's _content_ (objects are immutable and content-addressed, so
39
+
positive listings never go wrong), and the window bounds the negative-staleness risk.
40
+
Operators tune it with `-s3-cache-ttl`; `-s3-cache-ttl 0` disables the cache and restores
41
+
today's exact behavior.
42
+
43
+
**Defaults:** on by default, tunable; single-process unless peers are configured.
44
+
45
+
## Correctness model
46
+
47
+
- **Cache key space is the canonical full S3 key.** `Chroot` returns a _new_ `*S3FS`
48
+
with a different `root`, so the `*ListingCache` is **shared by pointer** across a root
49
+
fs and all chroot children, and prefixes are full-canonical (`fs3.key()`/`cleanPath` —
50
+
root-joined, leading slash stripped). `chroot.go:21` must copy the cache pointer.
51
+
- **groupcache key** = `prefix "\x00" window "\x00" localgen`, where
52
+
`window = now.Unix() / int64(ttl.Seconds())` and `localgen = c.gen(prefix)`. The
53
+
getter parses `prefix` back off the key (segment before the first `\x00`) and lists it.
54
+
- **Population is via the getter only** (groupcache has no `Set`). Both `ReadDir` and the
55
+
`Stat`/`Open` resolution route through `group.Get`, so each folder is listed once per
56
+
(window, localgen) and the result feeds both. groupcache's singleflight dedupes
57
+
concurrent identical `Get`s (one list even when a clone fans out across siblings), and
58
+
the owning peer dedupes across the fleet.
59
+
- **Listing payload** carries enough to serve both consumers without a second call:
60
+
per child `{name, kind(file|dir), size, mtimeUnixNano}` (`CommonPrefixes`→dir with
61
+
zero size/mtime, `Contents`→file; file wins on the pathological file+prefix collision,
62
+
matching today's Head-first precedence). Encoded compactly (JSON is fine; listings are
63
+
small).
64
+
- **In-process writes bump the local generation** of the parent prefix at _write
65
+
completion_, moving the key so the next read re-lists:
66
+
- `s3WriteFile.Close` / `s3MultipartUploadFile.Close` — after the upload succeeds.
67
+
- `Rename` (covers `TempFile`→final pack promotion), `Remove`, `MkdirAll`.
68
+
- **Positive `Stat`/`Open` of a file** is served from a second cache, the **head
69
+
cache** (see [Head precache](#head-precache)), rather than a foreground `HeadObject`.
70
+
Listings omit the user-metadata the unix-metadata feature needs, so head metadata is
71
+
cached per object and **prefetched in the background** when a listing is filled. `Open`
72
+
still issues `GetObject` for the body but skips its `HeadObject`. A `NoSuchKey` from a
73
+
delete racing the listing maps to `NotExist`.
74
+
75
+
## Changes
76
+
77
+
### go.mod
78
+
79
+
- Add `github.com/golang/groupcache` (effectively dependency-free).
80
+
81
+
### New file: `internal/s3fs/listingcache.go`
82
+
83
+
The cache wrapper, groupcache group, key logic, and background warmer.
84
+
85
+
```go
86
+
type CacheConfig struct {
87
+
TTL, RefreshInterval, IdleTTL time.Duration
88
+
SizeBytes int64 // groupcache LRU budget
89
+
Name string // group name (default "objgit-listings")
90
+
DisableHeadPrefetch bool // zero value = background head prefetch on
91
+
Self string // this node's groupcache URL ("" = single-process)
92
+
Peers []string // peer URLs (incl. self) when sharing
93
+
}
94
+
95
+
type childKind uint8 // kindFile / kindDir
96
+
type childEntry struct { Name string; Kind childKind; Size, Mtime int64 }
97
+
type headData struct { Size, Mtime int64; Meta map[string]string }
98
+
99
+
type ListingCache struct {
100
+
group *groupcache.Group // listings, keyed by prefix
101
+
headGroup *groupcache.Group // HeadObject metadata, keyed by object key
102
+
pool *groupcache.HTTPPool // nil in single-process mode
103
+
ttl time.Duration
104
+
cfg CacheConfig
105
+
clock func() time.Time // overridable in tests
106
+
prefetchSem chan struct{} // bounds background head precaches
107
+
mu sync.Mutex
108
+
gens map[string]uint64 // per-prefix local generation
109
+
seen map[string]time.Time // prefixes accessed → driven by the warmer
110
+
}
111
+
```
112
+
113
+
- `NewListingCache(cfg, client, bucket, separator) *ListingCache` — creates the
114
+
groupcache `Group` (name `"objgit-listings"`, `cfg.SizeBytes`) with a `GetterFunc`
115
+
closure over `client`/`bucket`/`separator` that parses the prefix from the key, runs a
116
+
full paginated `listChildren`, and encodes the payload into the sink. When
117
+
`cfg.Self != ""`, builds a `groupcache.NewHTTPPoolOpts(cfg.Self, …)` and `pool.Set(cfg.Peers…)`.
118
+
- `list(ctx, prefix) ([]childEntry, error)` — record `seen[prefix]=now`; build the
119
+
groupcache key; `group.Get(ctx, key, AllocatingByteSliceSink(&buf))`; decode. This is
120
+
the one entry point both `ReadDir` and `Stat`/`Open` use.
121
+
- `gen(prefix)` / `invalidate(prefix)` — `invalidate` bumps `gens[prefix]` and records
122
+
`seen` so the warmer re-warms the new key (no groupcache delete needed).
123
+
- `runWarmer(ctx)` / `RunWarmer(ctx)` — `time.NewTicker(RefreshInterval)`; each tick drop
124
+
`seen` entries idle past `IdleTTL`, then `group.Get` the current key for each remaining
125
+
prefix (pre-fills the new window before clients wait and smooths window-rollover
126
+
herds). No-op when `RefreshInterval<=0`. Returns on `ctx.Done()` (errgroup idiom).
127
+
- `PoolHandler() http.Handler` / `Stats()` accessors for `main` to serve peers and export
128
+
metrics.
129
+
- `splitKey(key) (prefix, base)` — split on last `/`; no slash → `("", key)`.
130
+
131
+
### Head precache
132
+
133
+
A **second groupcache group** (`<name>-heads`, keyed by object key) caches each file's
134
+
`HeadObject` result — `headData{size, mtimeUnixNano, meta}` — so positive `Stat`/`Open`
135
+
never pay a foreground `HeadObject`. The head key reuses the window + the **parent
136
+
prefix's** local generation, so a write under that prefix re-heads the object exactly as
137
+
it re-lists the folder.
138
+
139
+
- **Background prefetch (`prefetchHeads`).** When the listing getter fills a folder, it
140
+
launches a **detached goroutine per file** (`go headGroup.Get(context.Background(), …)`)
141
+
to warm the head cache off the request's critical path; groupcache singleflight
142
+
coalesces a background precache with any foreground head lookup for the same object. A
143
+
semaphore (`maxPrefetchInFlight = 64`) bounds the in-flight precaches so a large folder
144
+
can't spawn an unbounded goroutine/S3 storm — overflow files are fetched on demand
145
+
(logged at debug). `CacheConfig.DisableHeadPrefetch` turns this off (zero value = on).
146
+
- **`headInfo(ctx, key) (*s3.HeadObjectOutput, error)`** serves a foreground lookup from
147
+
the head cache (a warm hit costs no round-trip; a miss fills via one `HeadObject`).
148
+
- `Stat`/`Open` of a present file call `headInfo`; `newS3ReadFile` gains an optional
149
+
precomputed `*s3.HeadObjectOutput` so it skips its own `HeadObject` and only `GetObject`s
150
+
the body.
151
+
152
+
### `internal/s3fs/filesystem.go`
153
+
154
+
- Add `cache *ListingCache` to `S3FS`; `WithListingCache(c) Option`. `NewS3FS`'s `client`
155
+
parameter widens to an `s3Client` interface (the concrete `*storage.Client` satisfies
156
+
it) so tests can substitute a counting stub.
157
+
158
+
### `internal/s3fs/chroot.go`
159
+
160
+
- Copy `cache: fs3.cache` into the new `S3FS` literal (`chroot.go:21`).
161
+
162
+
### `internal/s3fs/dir.go`
163
+
164
+
- Extract `listChildren(ctx, client, bucket, separator, prefix) ([]childEntry, error)` —
165
+
the paginated `ListObjectsV2` loop, classifying `CommonPrefixes`→dir, `Contents`→file
166
+
with size/mtime, dirs-then-files preserving S3 order. A free function so the getter
167
+
(which holds only the raw client) can reuse it. Used by `ReadDir` (cache off) and both
168
+
getters.
169
+
- `ReadDir`: when `cache != nil`, get the entries via `cache.list(ctx, prefix)` and
170
+
build `[]fs.DirEntry` from them (rebuilding `newDirInfo`/`newFileInfo` from the payload);
171
+
otherwise list directly as today.
172
+
- `MkdirAll` — `cache.invalidate(parent prefix of filename)` after the `PutObject`.
173
+
174
+
### `internal/s3fs/basic.go`
175
+
176
+
- Helper `resolve(ctx, key) (childEntry, found, known bool)`: `(_,_,false)` when
177
+
`cache==nil`; else `prefix,base := splitKey(key)`, `entries, err := cache.list(...)`; on
178
+
error `known=false` (fall back to the live path — cache problems never fail the op);
179
+
else scan `entries` for `base` and return it.
180
+
- `Stat`: after the temp-buffer check, call `resolve`. If `known`: absent →
181
+
`&os.PathError{Op:"stat",…,Err: fs.ErrNotExist}`; dir → `newDirInfo`; file →
182
+
`cache.headInfo`→`newFileInfoFromHead` (head `NoSuchKey` → `NotExist`). Not `known` → the
183
+
existing `HeadObject`+probe fallback.
184
+
- `OpenFile` `O_RDONLY` (after temp check): same `resolve`; absent → `NotExist`, dir →
185
+
`newS3DirFile`, file → `cache.headInfo` then `newS3ReadFile(…, ho)` (skips its
186
+
`HeadObject`); not `known` → existing fallback with a nil head.
187
+
- `Rename` — invalidate parent prefix of both `src` and `dst` on success.
188
+
- `Remove` — invalidate parent prefix of `key` on success.
189
+
190
+
### `internal/s3fs/file.go`
191
+
192
+
- Add a `cache *ListingCache` field + constructor arg to `s3WriteFile` /
193
+
`s3MultipartUploadFile`; their `Close` calls `cache.invalidate(parent prefix of f.key)`
194
+
after a successful upload (nil-guarded). Update the two `OpenFile` call sites
195
+
(`basic.go:114`,`:117`).
196
+
197
+
### `internal/metrics/metrics.go`
198
+
199
+
- A Prometheus collector that reads `ListingCache.Stats()` (groupcache `group.Stats`:
200
+
Gets, CacheHits, Loads, LocalLoads, PeerLoads, LoaderErrors; plus `CacheStats` bytes/
201
+
items/evictions) and exports them as `objgit_s3_listing_cache_*` gauges/counters. No
202
+
`repo` label. (groupcache fills are already counted as `ListObjectsV2` via `observeS3`.)
203
+
`main` registers it only when the cache is enabled.
204
+
205
+
### `cmd/objgitd/main.go`
206
+
207
+
- New flags (kebab-case + flagenv) in the `var (...)` block at `main.go:32`:
208
+
- `-s3-cache-ttl` (`Duration`, default `60s`) — window size; `<=0` disables the cache.
209
+
- `-s3-cache-refresh` (`Duration`, default `30s`) — warmer interval; `<=0` disables the
210
+
warmer (lazy fill still works).
211
+
- `-s3-cache-idle` (`Duration`, default `10m`) — drop un-accessed prefixes from the warmer.
212
+
- `-s3-cache-size` (`Int64`, default `64<<20`) — groupcache LRU budget.
213
+
- `-groupcache-self` (`String`, default `""`) — this node's groupcache base URL; empty =
214
+
single-process.
215
+
- `-groupcache-peers` (`String`, default `""`) — comma-separated peer URLs.
216
+
- `-groupcache-bind` (`String`, default `""`) — listen addr serving `PoolHandler()`.
217
+
- When `ttl > 0`: build `cache := s3fs.NewListingCache(cfg, client, *bucket, "/")`, pass
218
+
`s3fs.WithListingCache(cache)` into `NewS3FS` (`main.go:78`), register the metrics
219
+
collector, and in the errgroup:
220
+
- if `-groupcache-bind != ""`, `net.Listen` + serve `cache.PoolHandler()` (plus the
221
+
Serve/`Shutdown`-on-`gCtx.Done()` two-goroutine idiom used by the other listeners).
222
+
- `g.Go(func() error { cache.RunWarmer(gCtx); return nil })`.
223
+
- Add the cache settings to the startup `slog.Info` line.
224
+
225
+
## Why this is acceptably safe
226
+
227
+
Within a single git operation the repeated negative lookups happen within milliseconds,
228
+
so even a 60s window eliminates essentially all redundant `HeadObject`+probe pairs, and
229
+
the first miss in a folder warms every sibling via one parent listing (deduped by
230
+
groupcache singleflight, fleet-wide via the owning peer). Local writes bump the
231
+
per-prefix generation, so a push reads its own objects immediately. git object _content_
232
+
is immutable, so positive/`ReadDir` results are never wrong about what exists — only the
233
+
_recency_ of newly-added entries is window-bounded. The residual risk is a cross-process
234
+
(or post-restart) negative read of an object another writer just created, bounded by one
235
+
TTL window — the limitation the user explicitly accepted; `-s3-cache-ttl 0` opts out
236
+
entirely. Note: writes fragment the shared keyspace for written prefixes (each writer's
237
+
`localgen` differs), so cross-process sharing is strongest on the read-only prefixes that
238
+
dominate clone/fetch and weakest on actively-pushed prefixes — which is the desirable
239
+
bias.
240
+
241
+
## Verification
242
+
243
+
1. `go build ./...`; `go mod tidy`; `go test ./...`.
244
+
2. **Cache disabled = no behavior change:** the cache is only wired when `-s3-cache-ttl>0`;
245
+
existing protocol tests (`go test ./cmd/objgitd/...`, needs `git` on PATH) must pass
246
+
with the cache off and on (single-process, no peers).
247
+
3. **New unit tests in `internal/s3fs`** (table-driven `tt`, counting-stub `storage.Client`):
248
+
- **Populate-on-miss:** first `Stat` of an absent key in a never-listed folder issues
249
+
exactly one `ListObjectsV2` (the parent) and **zero** `HeadObject`; a second absent
250
+
sibling → **zero** S3; a present sibling → only `HeadObject`.
251
+
- `ReadDir` then `Stat`/`Open` of an absent sibling → zero S3; a dir child → zero S3.
252
+
- **Local invalidation:** `Create`+`Close` / `Rename` / `Remove` / `MkdirAll` bump the
253
+
generation so a following `Stat`/`ReadDir` re-lists and sees the change (read-after-write).
254
+
- **Window TTL:** advancing time past `TTL` changes the key → re-list (inject a clock or
255
+
a settable `now` in `ListingCache` for the test).
256
+
- **Warmer:** `RunWarmer` re-`Get`s accessed prefixes and evicts idle ones past `IdleTTL`.
257
+
- **Chroot sharing:** `ReadDir` on the root then `Stat` of an absent child through a
258
+
chroot resolves from the same cached prefix (same canonical key).
259
+
- **Head precache:** one listing fill warms every file's head in the background; once
260
+
warm, `Stat` of a file does **zero** further `HeadObject`s (counting-stub tests
261
+
disable prefetch for determinism; a dedicated test enables it and waits for warmup).
262
+
4. **Two-process sharing (manual):** run two `objgitd` with `-groupcache-self`/`-peers`
263
+
pointing at each other; clone through one and confirm via `objgit_s3_listing_cache_*`
264
+
(PeerLoads > 0 on the non-owner) that listings are served cross-process.
265
+
5. **End-to-end:** `./objgitd -bucket $BUCKET -allow-push`; clone a packed repo twice and
266
+
confirm `objgit_s3_requests_total{operation="HeadObject"}` grows far slower than with
267
+
`-s3-cache-ttl 0`.
+2
go.mod
+2
go.mod
···
10
10
github.com/gliderlabs/ssh v0.3.8
11
11
github.com/go-git/go-billy/v6 v6.0.0-alpha.1
12
12
github.com/go-git/go-git/v6 v6.0.0-alpha.4
13
+
github.com/golang/groupcache v0.0.0-20241129210726-2c02b8208cf8
13
14
github.com/joho/godotenv v1.5.1
14
15
github.com/prometheus/client_golang v1.23.2
15
16
github.com/tigrisdata/storage-go v0.6.0
···
47
48
github.com/facebookgo/stack v0.0.0-20160209184415-751773369052 // indirect
48
49
github.com/facebookgo/subset v0.0.0-20200203212716-c811ad88dec4 // indirect
49
50
github.com/go-git/gcfg/v2 v2.0.2 // indirect
51
+
github.com/golang/protobuf v1.5.4 // indirect
50
52
github.com/kevinburke/ssh_config v1.6.0 // indirect
51
53
github.com/klauspost/cpuid/v2 v2.3.0 // indirect
52
54
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
+4
go.sum
+4
go.sum
···
75
75
github.com/go-git/go-git/v6 v6.0.0-alpha.4/go.mod h1:4ODa/G7hPWrh4Y+7lmt59Ij3zW38IEfvRoAZxLYYBhc=
76
76
github.com/go-quicktest/qt v1.101.0 h1:O1K29Txy5P2OK0dGo59b7b0LR6wKfIhttaAhHUyn7eI=
77
77
github.com/go-quicktest/qt v1.101.0/go.mod h1:14Bz/f7NwaXPtdYEgzsx46kqSxVwTbzVZsDC26tQJow=
78
+
github.com/golang/groupcache v0.0.0-20241129210726-2c02b8208cf8 h1:f+oWsMOmNPc8JmEHVZIycC7hBoQxHH9pNKQORJNozsQ=
79
+
github.com/golang/groupcache v0.0.0-20241129210726-2c02b8208cf8/go.mod h1:wcDNUvekVysuuOpQKo3191zZyTpiI6se1N1ULghS0sw=
80
+
github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
81
+
github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
78
82
github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
79
83
github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
80
84
github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0=
+60
internal/metrics/metrics.go
+60
internal/metrics/metrics.go
···
155
155
func ReposCreated() {
156
156
reposCreated.Inc()
157
157
}
158
+
159
+
// ListingCacheStats is a flat snapshot of the s3fs directory-listing cache's
160
+
// groupcache counters. It mirrors s3fs.CacheStats but is defined here so s3fs
161
+
// stays free of any Prometheus import; main bridges the two.
162
+
type ListingCacheStats struct {
163
+
Gets, CacheHits, Loads, LocalLoads, PeerLoads, LocalLoadErrs int64
164
+
MainBytes, MainItems, MainEvictions int64
165
+
HotBytes, HotItems int64
166
+
}
167
+
168
+
// RegisterListingCache installs a Prometheus collector that reports the
169
+
// directory-listing cache's groupcache counters under objgit_s3_listing_cache_*.
170
+
// provider is polled at scrape time. Call once at startup when the cache is
171
+
// enabled.
172
+
func RegisterListingCache(provider func() ListingCacheStats) {
173
+
prometheus.MustRegister(&listingCacheCollector{provider: provider})
174
+
}
175
+
176
+
type listingCacheCollector struct {
177
+
provider func() ListingCacheStats
178
+
}
179
+
180
+
var (
181
+
lcGets = prometheus.NewDesc("objgit_s3_listing_cache_gets_total", "Listing-cache Get requests (incl. from peers).", nil, nil)
182
+
lcHits = prometheus.NewDesc("objgit_s3_listing_cache_hits_total", "Listing-cache requests served from cache.", nil, nil)
183
+
lcLoads = prometheus.NewDesc("objgit_s3_listing_cache_loads_total", "Listing-cache loads (gets minus hits).", nil, nil)
184
+
lcLocal = prometheus.NewDesc("objgit_s3_listing_cache_local_loads_total", "Listing-cache loads served by listing S3 locally.", nil, nil)
185
+
lcPeer = prometheus.NewDesc("objgit_s3_listing_cache_peer_loads_total", "Listing-cache loads served by a peer.", nil, nil)
186
+
lcLoadErrs = prometheus.NewDesc("objgit_s3_listing_cache_local_load_errors_total", "Listing-cache local load errors.", nil, nil)
187
+
lcBytes = prometheus.NewDesc("objgit_s3_listing_cache_bytes", "Listing-cache resident bytes by cache.", []string{"cache"}, nil)
188
+
lcItems = prometheus.NewDesc("objgit_s3_listing_cache_items", "Listing-cache resident items by cache.", []string{"cache"}, nil)
189
+
lcEvictions = prometheus.NewDesc("objgit_s3_listing_cache_evictions_total", "Listing-cache main-cache evictions.", nil, nil)
190
+
)
191
+
192
+
func (c *listingCacheCollector) Describe(ch chan<- *prometheus.Desc) {
193
+
ch <- lcGets
194
+
ch <- lcHits
195
+
ch <- lcLoads
196
+
ch <- lcLocal
197
+
ch <- lcPeer
198
+
ch <- lcLoadErrs
199
+
ch <- lcBytes
200
+
ch <- lcItems
201
+
ch <- lcEvictions
202
+
}
203
+
204
+
func (c *listingCacheCollector) Collect(ch chan<- prometheus.Metric) {
205
+
s := c.provider()
206
+
ch <- prometheus.MustNewConstMetric(lcGets, prometheus.CounterValue, float64(s.Gets))
207
+
ch <- prometheus.MustNewConstMetric(lcHits, prometheus.CounterValue, float64(s.CacheHits))
208
+
ch <- prometheus.MustNewConstMetric(lcLoads, prometheus.CounterValue, float64(s.Loads))
209
+
ch <- prometheus.MustNewConstMetric(lcLocal, prometheus.CounterValue, float64(s.LocalLoads))
210
+
ch <- prometheus.MustNewConstMetric(lcPeer, prometheus.CounterValue, float64(s.PeerLoads))
211
+
ch <- prometheus.MustNewConstMetric(lcLoadErrs, prometheus.CounterValue, float64(s.LocalLoadErrs))
212
+
ch <- prometheus.MustNewConstMetric(lcEvictions, prometheus.CounterValue, float64(s.MainEvictions))
213
+
ch <- prometheus.MustNewConstMetric(lcBytes, prometheus.GaugeValue, float64(s.MainBytes), "main")
214
+
ch <- prometheus.MustNewConstMetric(lcBytes, prometheus.GaugeValue, float64(s.HotBytes), "hot")
215
+
ch <- prometheus.MustNewConstMetric(lcItems, prometheus.GaugeValue, float64(s.MainItems), "main")
216
+
ch <- prometheus.MustNewConstMetric(lcItems, prometheus.GaugeValue, float64(s.HotItems), "hot")
217
+
}
+79
-3
internal/s3fs/basic.go
+79
-3
internal/s3fs/basic.go
···
74
74
return &tempReadFile{buf: buf, name: filename}, nil
75
75
}
76
76
77
-
f, err := newS3ReadFile(fs3.client, fs3.bucket, key, filename)
77
+
// If the parent folder's listing is cached, resolve the open without a
78
+
// negotiation round-trip: absent → not-exist, a sub-prefix → directory.
79
+
// For a present file the head cache (prewarmed in the background) lets
80
+
// newS3ReadFile skip its HeadObject; only GetObject fetches the body.
81
+
if info, found, known := fs3.resolve(context.TODO(), key); known {
82
+
switch {
83
+
case !found:
84
+
return nil, &os.PathError{Op: "open", Path: filename, Err: fs.ErrNotExist}
85
+
case info.Kind == kindDir:
86
+
return newS3DirFile(key, fs3.bucket, fs3.client), nil
87
+
default:
88
+
if ho, err := fs3.cache.headInfo(context.TODO(), key); err == nil {
89
+
return newS3ReadFile(fs3.client, fs3.bucket, key, filename, ho)
90
+
} else if isNotFound(err) {
91
+
return nil, &os.PathError{Op: "open", Path: filename, Err: fs.ErrNotExist}
92
+
}
93
+
// transient head-cache error: fall through to a live read.
94
+
}
95
+
}
96
+
97
+
f, err := newS3ReadFile(fs3.client, fs3.bucket, key, filename, nil)
78
98
if err == nil {
79
99
return f, nil
80
100
}
···
111
131
return nil, &os.PathError{Op: "open", Path: filename, Err: fs.ErrNotExist}
112
132
113
133
case O_WRONLY:
114
-
return newS3WriteFile(fs3.client, fs3.bucket, key, filename, fs3.unixMeta)
134
+
return newS3WriteFile(fs3.client, fs3.bucket, key, filename, fs3.unixMeta, fs3.cache)
115
135
116
136
case O_WRMULTIPART:
117
-
return newS3MultipartUploadFile(fs3.client, fs3.bucket, key, filename, fs3.unixMeta)
137
+
return newS3MultipartUploadFile(fs3.client, fs3.bucket, key, filename, fs3.unixMeta, fs3.cache)
118
138
119
139
default:
120
140
return nil, errors.New("unsupported open flag")
121
141
}
122
142
}
123
143
144
+
// resolve consults the listing cache for key's parent folder. known is false
145
+
// when the cache is disabled or the lookup failed — the caller then falls back
146
+
// to a live S3 round-trip, so cache problems never fail an operation. When
147
+
// known, found reports whether base exists under the parent and info describes
148
+
// it (kind/size/mtime).
149
+
func (fs3 *S3FS) resolve(ctx context.Context, key string) (info childEntry, found, known bool) {
150
+
if fs3.cache == nil {
151
+
return childEntry{}, false, false
152
+
}
153
+
prefix, base := splitKey(key)
154
+
entries, err := fs3.cache.list(ctx, prefix)
155
+
if err != nil {
156
+
return childEntry{}, false, false
157
+
}
158
+
for _, e := range entries {
159
+
if e.Name == base {
160
+
return e, true, true
161
+
}
162
+
}
163
+
return childEntry{}, false, true
164
+
}
165
+
124
166
// Stat returns a FileInfo describing the named file.
125
167
func (fs3 *S3FS) Stat(filename string) (os.FileInfo, error) {
126
168
key := strings.TrimPrefix(fs3.cleanPath(filename), "/")
···
136
178
137
179
ctx := context.TODO()
138
180
181
+
// If the parent folder's listing is cached, answer without a foreground S3
182
+
// round-trip: absent → not-exist, a sub-prefix → directory, a present file
183
+
// → the head cache (prewarmed in the background from the listing).
184
+
if info, found, known := fs3.resolve(ctx, key); known {
185
+
if !found {
186
+
return nil, &os.PathError{Op: "stat", Path: filename, Err: fs.ErrNotExist}
187
+
}
188
+
if info.Kind == kindDir {
189
+
return newDirInfo(path.Base(key)), nil
190
+
}
191
+
if ho, err := fs3.cache.headInfo(ctx, key); err == nil {
192
+
return newFileInfoFromHead(path.Base(key), ho, fs3.unixMeta), nil
193
+
} else if isNotFound(err) {
194
+
return nil, &os.PathError{Op: "stat", Path: filename, Err: fs.ErrNotExist}
195
+
}
196
+
// transient head-cache error: fall through to a direct HeadObject.
197
+
}
198
+
139
199
start := time.Now()
140
200
head, err := fs3.client.HeadObject(ctx, &s3.HeadObjectInput{
141
201
Bucket: &fs3.bucket,
···
199
259
if err != nil {
200
260
return fmt.Errorf("failed to upload temp %q to %q: %w", oldpath, newpath, err)
201
261
}
262
+
if fs3.cache != nil {
263
+
prefix, _ := splitKey(dst)
264
+
fs3.cache.invalidate(prefix)
265
+
}
202
266
return nil
203
267
}
204
268
···
217
281
return fmt.Errorf("failed to rename %q to %q: %w", oldpath, newpath, err)
218
282
}
219
283
284
+
// Both folders changed: the source lost a child, the destination gained one.
285
+
if fs3.cache != nil {
286
+
srcPrefix, _ := splitKey(src)
287
+
dstPrefix, _ := splitKey(dst)
288
+
fs3.cache.invalidate(srcPrefix)
289
+
fs3.cache.invalidate(dstPrefix)
290
+
}
291
+
220
292
return nil
221
293
}
222
294
···
241
313
observeS3("DeleteObject", start, err)
242
314
if err != nil {
243
315
return fmt.Errorf("failed to remove file: %w", err)
316
+
}
317
+
if fs3.cache != nil {
318
+
prefix, _ := splitKey(key)
319
+
fs3.cache.invalidate(prefix)
244
320
}
245
321
return nil
246
322
}
+1
internal/s3fs/chroot.go
+1
internal/s3fs/chroot.go
+57
-24
internal/s3fs/dir.go
+57
-24
internal/s3fs/dir.go
···
15
15
"github.com/aws/aws-sdk-go-v2/service/s3"
16
16
)
17
17
18
-
// ReadDir reads the directory named by dirname and returns a list of
19
-
// directory entries sorted by filename.
20
-
func (fs3 *S3FS) ReadDir(dir string) ([]fs.DirEntry, error) {
21
-
key := strings.TrimPrefix(fs3.cleanPath(dir), "/")
22
-
var prefix string
23
-
if key != "" && key != "." {
24
-
prefix = key + "/"
25
-
}
26
-
27
-
ctx := context.TODO()
28
-
18
+
// listChildren lists the immediate children of prefix (a full-canonical key
19
+
// prefix: "" for the bucket root, otherwise ending in separator), paginating to
20
+
// completion. Sub-prefixes come back as kindDir, objects as kindFile carrying
21
+
// size/mtime — dirs first then files, preserving S3's lexicographic order. It
22
+
// is a free function so the listing cache's getter, which holds only the raw
23
+
// client, can reuse it.
24
+
func listChildren(ctx context.Context, client s3Client, bucket, separator, prefix string) ([]childEntry, error) {
29
25
var ct *string
30
-
var dirs []fs.DirEntry
31
-
var files []fs.DirEntry
26
+
var dirs, files []childEntry
32
27
for {
33
28
start := time.Now()
34
-
res, err := fs3.client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
35
-
Bucket: &fs3.bucket,
29
+
res, err := client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
30
+
Bucket: &bucket,
36
31
Prefix: &prefix,
37
32
ContinuationToken: ct,
38
-
Delimiter: &fs3.separator,
33
+
Delimiter: &separator,
39
34
})
40
35
observeS3("ListObjectsV2", start, err)
41
36
if err != nil {
···
47
42
if name == "" {
48
43
continue
49
44
}
50
-
dirs = append(dirs, fs.FileInfoToDirEntry(newDirInfo(name)))
45
+
dirs = append(dirs, childEntry{Name: name, Kind: kindDir})
51
46
}
52
47
53
48
for _, f := range res.Contents {
···
60
55
if name == "" {
61
56
continue
62
57
}
63
-
files = append(files,
64
-
fs.FileInfoToDirEntry(newFileInfo(
65
-
pathpkg.Base(name),
66
-
aws.ToInt64(f.Size),
67
-
aws.ToTime(f.LastModified),
68
-
)),
69
-
)
58
+
files = append(files, childEntry{
59
+
Name: pathpkg.Base(name),
60
+
Kind: kindFile,
61
+
Size: aws.ToInt64(f.Size),
62
+
Mtime: aws.ToTime(f.LastModified).UnixNano(),
63
+
})
70
64
}
71
65
72
66
if !aws.ToBool(res.IsTruncated) {
···
78
72
return append(dirs, files...), nil
79
73
}
80
74
75
+
// ReadDir reads the directory named by dirname and returns a list of
76
+
// directory entries. When a listing cache is attached the listing is served
77
+
// through it (and reused by later Stat/Open of siblings).
78
+
func (fs3 *S3FS) ReadDir(dir string) ([]fs.DirEntry, error) {
79
+
key := strings.TrimPrefix(fs3.cleanPath(dir), "/")
80
+
var prefix string
81
+
if key != "" && key != "." {
82
+
prefix = key + "/"
83
+
}
84
+
85
+
ctx := context.TODO()
86
+
87
+
var entries []childEntry
88
+
var err error
89
+
if fs3.cache != nil {
90
+
entries, err = fs3.cache.list(ctx, prefix)
91
+
} else {
92
+
entries, err = listChildren(ctx, fs3.client, fs3.bucket, fs3.separator, prefix)
93
+
}
94
+
if err != nil {
95
+
return nil, err
96
+
}
97
+
98
+
out := make([]fs.DirEntry, 0, len(entries))
99
+
for _, e := range entries {
100
+
if e.Kind == kindDir {
101
+
out = append(out, fs.FileInfoToDirEntry(newDirInfo(e.Name)))
102
+
continue
103
+
}
104
+
out = append(out, fs.FileInfoToDirEntry(newFileInfo(e.Name, e.Size, time.Unix(0, e.Mtime))))
105
+
}
106
+
107
+
return out, nil
108
+
}
109
+
81
110
// MkdirAll creates a directory named path, along with any necessary
82
111
// parents, and returns nil, or else returns an error. The permission bits
83
112
// perm are used for all directories that MkdirAll creates. If path is/
···
90
119
Body: bytes.NewBuffer(nil),
91
120
})
92
121
observeS3("PutObject", start, err)
122
+
if err == nil && fs3.cache != nil {
123
+
prefix, _ := splitKey(fs3.key(filename))
124
+
fs3.cache.invalidate(prefix)
125
+
}
93
126
94
127
return err
95
128
}
+47
-26
internal/s3fs/file.go
+47
-26
internal/s3fs/file.go
···
13
13
"time"
14
14
15
15
"github.com/aws/aws-sdk-go-v2/service/s3"
16
-
"github.com/tigrisdata/storage-go"
17
16
"go.uber.org/atomic"
18
17
"tangled.org/xeiaso.net/objgit/internal/s3fs/unixmeta"
19
18
)
···
50
49
//
51
50
// Upon creation, the file is loaded from S3.
52
51
type s3ReadFile struct {
53
-
client *storage.Client // S3 SDK client
52
+
client s3Client // S3 SDK client
54
53
bucket string // S3 bucket name
55
54
key string // File object's key in S3
56
55
name string // Root-relative path as presented to Open
···
60
59
}
61
60
62
61
// newS3ReadFile creates a new s3ReadFile. key is the full S3 object key; name
63
-
// is the root-relative path the caller passed to Open (returned by Name).
64
-
func newS3ReadFile(client *storage.Client, bucket, key, name string) (*s3ReadFile, error) {
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.
66
+
func newS3ReadFile(client s3Client, bucket, key, name string, head *s3.HeadObjectOutput) (*s3ReadFile, error) {
65
67
// Create the context
66
68
ctx := context.TODO() // TODO: How can user-supplied contexts be supported?
67
69
68
-
start := time.Now()
69
-
ho, err := client.HeadObject(ctx, &s3.HeadObjectInput{
70
-
Bucket: new(bucket),
71
-
Key: new(key),
72
-
})
73
-
observeS3("HeadObject", start, err)
74
-
if err != nil {
75
-
return nil, &os.PathError{Op: "read", Path: key, Err: err}
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
76
81
}
77
82
78
83
// Run the GetObject operation
79
-
start = time.Now()
84
+
start := time.Now()
80
85
res, err := client.GetObject(ctx, &s3.GetObjectInput{
81
86
Bucket: &bucket,
82
87
Key: &key,
···
100
105
key: key,
101
106
name: name,
102
107
reader: reader,
103
-
head: ho,
108
+
head: head,
104
109
}, nil
105
110
}
106
111
···
179
184
// Upon creation, a buffer is created to store the file contents. Upon close,
180
185
// the file is uploaded to S3.
181
186
type s3WriteFile struct {
182
-
client *storage.Client // s3 skd client
187
+
client s3Client // s3 skd client
183
188
bucket string // S3 bucket name
184
189
key string // File object's key in S3
185
190
name string // Root-relative path as presented to Open
186
191
closed bool // Is the file closed?
187
192
buf *bytes.Buffer // Buffer for storing the file before it's uploaded
188
193
unixMeta *unixMetaConfig // optional POSIX attribute defaults (nil = disabled)
194
+
cache *ListingCache // optional listing cache to invalidate on upload
189
195
}
190
196
191
197
// newS3WriteFile creates a new s3WriteFile. key is the full S3 object key; name
192
198
// is the root-relative path the caller passed to Open (returned by Name).
193
-
func newS3WriteFile(client *storage.Client, bucket, key, name string, cfg *unixMetaConfig) (*s3WriteFile, error) {
199
+
func newS3WriteFile(client s3Client, bucket, key, name string, cfg *unixMetaConfig, cache *ListingCache) (*s3WriteFile, error) {
194
200
return &s3WriteFile{
195
201
client: client,
196
202
bucket: bucket,
···
198
204
name: name,
199
205
buf: bytes.NewBuffer(nil),
200
206
unixMeta: cfg,
207
+
cache: cache,
201
208
}, nil
202
209
}
203
210
···
262
269
return fmt.Errorf("unable to perform GetObject operation: %w", err)
263
270
}
264
271
272
+
// The new object changes its parent folder's listing; drop the cached
273
+
// listing so the next read re-lists and sees it.
274
+
if f.cache != nil {
275
+
prefix, _ := splitKey(f.key)
276
+
f.cache.invalidate(prefix)
277
+
}
278
+
265
279
return nil
266
280
}
267
281
···
287
301
288
302
// s3MultipartUploadFile implements billy.File
289
303
type s3MultipartUploadFile struct {
290
-
client *storage.Client // s3 skd client
291
-
bucket string // S3 bucket name
292
-
key string // File object's key in S3
293
-
name string // Root-relative path as presented to Open
294
-
closed bool // Is the file closed?
295
-
uploadID string // S3 multipart upload ID
296
-
uploadN *atomic.Int32 // Counter tracking the number of uploads
304
+
client s3Client // s3 skd client
305
+
bucket string // S3 bucket name
306
+
key string // File object's key in S3
307
+
name string // Root-relative path as presented to Open
308
+
closed bool // Is the file closed?
309
+
uploadID string // S3 multipart upload ID
310
+
uploadN *atomic.Int32 // Counter tracking the number of uploads
311
+
cache *ListingCache // optional listing cache to invalidate on upload
297
312
}
298
313
299
314
// newS3MultipartUploadFile creates a new s3MultipartUploadFile. key is the full
300
315
// S3 object key; name is the root-relative path passed to Open.
301
-
func newS3MultipartUploadFile(client *storage.Client, bucket, key, name string, cfg *unixMetaConfig) (*s3MultipartUploadFile, error) {
316
+
func newS3MultipartUploadFile(client s3Client, bucket, key, name string, cfg *unixMetaConfig, cache *ListingCache) (*s3MultipartUploadFile, error) {
302
317
// TODO: Check if the file exists
303
318
// ...
304
319
···
326
341
name: name,
327
342
uploadID: *res.UploadId,
328
343
uploadN: atomic.NewInt32(1),
344
+
cache: cache,
329
345
}, nil
330
346
}
331
347
···
412
428
return fmt.Errorf("unable to complete multipart upload: %w", err)
413
429
}
414
430
431
+
if f.cache != nil {
432
+
prefix, _ := splitKey(f.key)
433
+
f.cache.invalidate(prefix)
434
+
}
435
+
415
436
return nil
416
437
}
417
438
···
430
451
type s3DirFile struct {
431
452
name, bucket string
432
453
closed bool
433
-
cli *storage.Client
454
+
cli s3Client
434
455
}
435
456
436
-
func newS3DirFile(name, bucket string, cli *storage.Client) *s3DirFile {
457
+
func newS3DirFile(name, bucket string, cli s3Client) *s3DirFile {
437
458
return &s3DirFile{
438
459
name: name,
439
460
bucket: bucket,
+35
-4
internal/s3fs/filesystem.go
+35
-4
internal/s3fs/filesystem.go
···
1
1
package s3fs
2
2
3
3
import (
4
+
"context"
4
5
"fmt"
5
6
"os"
6
7
"path"
7
8
"strings"
8
9
"sync"
9
10
11
+
"github.com/aws/aws-sdk-go-v2/service/s3"
10
12
"github.com/go-git/go-billy/v6"
11
-
"github.com/tigrisdata/storage-go"
12
13
)
13
14
14
15
const (
15
16
DefaultSeparator = "/"
16
17
)
17
18
19
+
// s3Client is the subset of *storage.Client (which embeds *s3.Client) that the
20
+
// filesystem uses. Naming it as an interface lets tests substitute a counting
21
+
// stub; the concrete Tigris client satisfies it unchanged.
22
+
type s3Client interface {
23
+
HeadObject(context.Context, *s3.HeadObjectInput, ...func(*s3.Options)) (*s3.HeadObjectOutput, error)
24
+
GetObject(context.Context, *s3.GetObjectInput, ...func(*s3.Options)) (*s3.GetObjectOutput, error)
25
+
PutObject(context.Context, *s3.PutObjectInput, ...func(*s3.Options)) (*s3.PutObjectOutput, error)
26
+
ListObjectsV2(context.Context, *s3.ListObjectsV2Input, ...func(*s3.Options)) (*s3.ListObjectsV2Output, error)
27
+
DeleteObject(context.Context, *s3.DeleteObjectInput, ...func(*s3.Options)) (*s3.DeleteObjectOutput, error)
28
+
RenameObject(context.Context, *s3.CopyObjectInput, ...func(*s3.Options)) (*s3.CopyObjectOutput, error)
29
+
CreateMultipartUpload(context.Context, *s3.CreateMultipartUploadInput, ...func(*s3.Options)) (*s3.CreateMultipartUploadOutput, error)
30
+
UploadPart(context.Context, *s3.UploadPartInput, ...func(*s3.Options)) (*s3.UploadPartOutput, error)
31
+
CompleteMultipartUpload(context.Context, *s3.CompleteMultipartUploadInput, ...func(*s3.Options)) (*s3.CompleteMultipartUploadOutput, error)
32
+
}
33
+
18
34
// unixMetaConfig holds the session defaults used when the optional Unix-metadata
19
35
// feature is enabled. A nil *unixMetaConfig means the feature is off and the
20
36
// filesystem behaves as if no POSIX attributes exist.
···
24
40
}
25
41
26
42
type S3FS struct {
27
-
client *storage.Client
43
+
client s3Client
28
44
bucket string
29
45
root string
30
46
separator string
31
47
unixMeta *unixMetaConfig
32
48
49
+
// cache, when non-nil, memoises directory listings so Stat/Open of a path
50
+
// whose parent folder has been listed can skip the S3 round-trip. It is
51
+
// shared by pointer across this filesystem and all of its Chroot children.
52
+
cache *ListingCache
53
+
33
54
// temps holds TempFile-backed buffers keyed by canonical S3 key, so a
34
55
// subsequent Open of the same path returns a reader over the same bytes
35
56
// the writer is still appending to. See tempfs.go.
···
51
72
}
52
73
}
53
74
54
-
// NewS3FS creates a new S3FS Filesystem.
55
-
func NewS3FS(client *storage.Client, bucket string, opts ...Option) (billy.Filesystem, error) {
75
+
// WithListingCache attaches a directory-listing cache. The same *ListingCache is
76
+
// carried into every Chroot child so the whole tree shares one cache keyed by
77
+
// full-canonical prefix. Construct the cache with NewListingCache.
78
+
func WithListingCache(c *ListingCache) Option {
79
+
return func(fs3 *S3FS) {
80
+
fs3.cache = c
81
+
}
82
+
}
83
+
84
+
// NewS3FS creates a new S3FS Filesystem. client is typically a *storage.Client;
85
+
// it is accepted as the s3Client interface so tests can substitute a stub.
86
+
func NewS3FS(client s3Client, bucket string, opts ...Option) (billy.Filesystem, error) {
56
87
// Check for a non-nil client
57
88
if client == nil {
58
89
return nil, fmt.Errorf("s3 client cannot be nil")
+392
internal/s3fs/listingcache.go
+392
internal/s3fs/listingcache.go
···
1
+
// listingcache.go caches directory listings so that Stat/Open of a path whose
2
+
// parent folder has been listed can be answered without an S3 round-trip: an
3
+
// absent child returns "not found" for free, a sub-directory is recognised for
4
+
// free, and only a present file still issues a HeadObject/GetObject for its
5
+
// authoritative content/metadata.
6
+
//
7
+
// The store is github.com/golang/groupcache so the listing can be shared across
8
+
// a fleet via its consistent-hash peer pool. groupcache is fill-once and
9
+
// immutable (no Set, Delete, or TTL), so two facts are encoded into the cache
10
+
// key instead:
11
+
//
12
+
// - a time window floor(now/TTL): when it advances the key changes, forcing a
13
+
// re-list. This bounds staleness from writers this process cannot see.
14
+
// - a per-prefix local generation, bumped on every local write under the
15
+
// prefix: this moves the key so the next local read re-lists and sees the
16
+
// write immediately (read-after-write correctness groupcache alone cannot
17
+
// give). The stale entry under the old key is simply never queried again.
18
+
package s3fs
19
+
20
+
import (
21
+
"context"
22
+
"encoding/json"
23
+
"errors"
24
+
"log/slog"
25
+
"net/http"
26
+
"strconv"
27
+
"strings"
28
+
"sync"
29
+
"time"
30
+
31
+
"github.com/aws/aws-sdk-go-v2/aws"
32
+
"github.com/aws/aws-sdk-go-v2/service/s3"
33
+
"github.com/aws/smithy-go"
34
+
"github.com/golang/groupcache"
35
+
)
36
+
37
+
// maxPrefetchInFlight bounds the background HeadObject precaches in flight at
38
+
// once, so listing a large folder can't spawn an unbounded goroutine/S3 storm.
39
+
// Overflow files are simply fetched on demand instead.
40
+
const maxPrefetchInFlight = 64
41
+
42
+
// childKind distinguishes the two things a listing can report under a prefix.
43
+
type childKind uint8
44
+
45
+
const (
46
+
kindFile childKind = iota
47
+
kindDir
48
+
)
49
+
50
+
// childEntry is one immediate child of a listed prefix. name is the child's
51
+
// base name (no separator). size/mtime are populated for files only; dirs
52
+
// (S3 common prefixes) carry zero values.
53
+
type childEntry struct {
54
+
Name string `json:"n"`
55
+
Kind childKind `json:"k"`
56
+
Size int64 `json:"s,omitempty"`
57
+
Mtime int64 `json:"m,omitempty"` // unix nanoseconds
58
+
}
59
+
60
+
// headData is the cached result of a HeadObject: everything the file metadata
61
+
// helpers need. Meta carries the x-amz-meta-* user metadata for the
62
+
// Unix-metadata feature (listings can't, which is why heads are cached too).
63
+
type headData struct {
64
+
Size int64 `json:"s"`
65
+
Mtime int64 `json:"m"` // unix nanoseconds
66
+
Meta map[string]string `json:"u,omitempty"`
67
+
}
68
+
69
+
// DefaultGroupName is the groupcache group every objgitd uses; it must match
70
+
// across peers for cross-process sharing to route correctly.
71
+
const DefaultGroupName = "objgit-listings"
72
+
73
+
// CacheConfig configures a ListingCache. TTL must be > 0 (callers gate the whole
74
+
// feature on it). The remaining fields have sane zero-value behaviour.
75
+
type CacheConfig struct {
76
+
TTL time.Duration // key window; bounds cross-process staleness
77
+
RefreshInterval time.Duration // warmer tick; <=0 disables the warmer
78
+
IdleTTL time.Duration // drop un-accessed prefixes from the warmer
79
+
SizeBytes int64 // groupcache LRU budget (<=0 → 64 MiB)
80
+
Name string // groupcache group name (default DefaultGroupName)
81
+
82
+
// DisableHeadPrefetch turns off the background HeadObject precache that
83
+
// warms the head cache for every file in a listing. The zero value keeps
84
+
// prefetching enabled.
85
+
DisableHeadPrefetch bool
86
+
87
+
Self string // this node's groupcache base URL ("" → single-process)
88
+
Peers []string // peer base URLs, including Self, when sharing
89
+
}
90
+
91
+
// ListingCache wraps a groupcache group with the window/generation key scheme
92
+
// and a background warmer. It is safe for concurrent use and is shared by
93
+
// pointer across an S3FS and all of its Chroot children.
94
+
type ListingCache struct {
95
+
group *groupcache.Group
96
+
headGroup *groupcache.Group // per-object HeadObject metadata
97
+
pool *groupcache.HTTPPool // nil in single-process mode
98
+
ttl time.Duration
99
+
cfg CacheConfig
100
+
101
+
clock func() time.Time // overridable in tests
102
+
prefetchSem chan struct{} // bounds background head precaches
103
+
104
+
mu sync.Mutex
105
+
gens map[string]uint64 // per-prefix local generation
106
+
seen map[string]time.Time // prefixes accessed → driven by the warmer
107
+
}
108
+
109
+
// NewListingCache builds a cache backed by a groupcache group whose getter lists
110
+
// the requested prefix from S3. client/bucket/separator are captured for the
111
+
// getter; they are the root filesystem's, so cached prefixes are full-canonical
112
+
// keys that every Chroot view agrees on.
113
+
func NewListingCache(cfg CacheConfig, client s3Client, bucket, separator string) *ListingCache {
114
+
if cfg.Name == "" {
115
+
cfg.Name = DefaultGroupName
116
+
}
117
+
if cfg.SizeBytes <= 0 {
118
+
cfg.SizeBytes = 64 << 20
119
+
}
120
+
c := &ListingCache{
121
+
ttl: cfg.TTL,
122
+
cfg: cfg,
123
+
clock: time.Now,
124
+
prefetchSem: make(chan struct{}, maxPrefetchInFlight),
125
+
gens: map[string]uint64{},
126
+
seen: map[string]time.Time{},
127
+
}
128
+
129
+
listGetter := groupcache.GetterFunc(func(ctx context.Context, key string, dest groupcache.Sink) error {
130
+
prefix := keyPrefix(key)
131
+
entries, err := listChildren(ctx, client, bucket, separator, prefix)
132
+
if err != nil {
133
+
return err
134
+
}
135
+
data, err := json.Marshal(entries)
136
+
if err != nil {
137
+
return err
138
+
}
139
+
// Precache each file's HeadObject in the background, off this request's
140
+
// critical path; groupcache singleflight coalesces a background precache
141
+
// with any foreground head lookup for the same object.
142
+
c.prefetchHeads(prefix, entries)
143
+
return dest.SetBytes(data)
144
+
})
145
+
c.group = groupcache.NewGroup(cfg.Name, cfg.SizeBytes, listGetter)
146
+
147
+
headGetter := groupcache.GetterFunc(func(ctx context.Context, key string, dest groupcache.Sink) error {
148
+
objKey := keyPrefix(key)
149
+
start := time.Now()
150
+
ho, err := client.HeadObject(ctx, &s3.HeadObjectInput{Bucket: &bucket, Key: &objKey})
151
+
observeS3("HeadObject", start, err)
152
+
if err != nil {
153
+
return err
154
+
}
155
+
data, err := json.Marshal(headData{
156
+
Size: aws.ToInt64(ho.ContentLength),
157
+
Mtime: aws.ToTime(ho.LastModified).UnixNano(),
158
+
Meta: ho.Metadata,
159
+
})
160
+
if err != nil {
161
+
return err
162
+
}
163
+
return dest.SetBytes(data)
164
+
})
165
+
c.headGroup = groupcache.NewGroup(cfg.Name+"-heads", cfg.SizeBytes, headGetter)
166
+
167
+
if cfg.Self != "" {
168
+
c.pool = groupcache.NewHTTPPoolOpts(cfg.Self, &groupcache.HTTPPoolOptions{})
169
+
peers := cfg.Peers
170
+
if len(peers) == 0 {
171
+
peers = []string{cfg.Self}
172
+
}
173
+
c.pool.Set(peers...)
174
+
}
175
+
176
+
return c
177
+
}
178
+
179
+
// list returns the immediate children of prefix, served from groupcache (which
180
+
// fills via S3 on a miss). prefix is a full-canonical key prefix: "" for the
181
+
// bucket root, otherwise ending in the separator.
182
+
func (c *ListingCache) list(ctx context.Context, prefix string) ([]childEntry, error) {
183
+
c.touch(prefix)
184
+
185
+
var data []byte
186
+
if err := c.group.Get(ctx, c.groupKey(prefix), groupcache.AllocatingByteSliceSink(&data)); err != nil {
187
+
return nil, err
188
+
}
189
+
var entries []childEntry
190
+
if err := json.Unmarshal(data, &entries); err != nil {
191
+
return nil, err
192
+
}
193
+
return entries, nil
194
+
}
195
+
196
+
// groupKey encodes the prefix, the current time window, and the prefix's local
197
+
// generation. The prefix is first so the getter can recover it.
198
+
func (c *ListingCache) groupKey(prefix string) string {
199
+
window := c.clock().UnixNano() / int64(c.ttl)
200
+
return prefix + "\x00" + strconv.FormatInt(window, 10) + "\x00" + strconv.FormatUint(c.gen(prefix), 10)
201
+
}
202
+
203
+
// headKey is groupKey's analogue for the per-object head cache. It carries the
204
+
// object key plus the same window and its parent prefix's generation, so a
205
+
// write under that prefix (which bumps the generation) re-heads the object.
206
+
func (c *ListingCache) headKey(objKey string) string {
207
+
prefix, _ := splitKey(objKey)
208
+
window := c.clock().UnixNano() / int64(c.ttl)
209
+
return objKey + "\x00" + strconv.FormatInt(window, 10) + "\x00" + strconv.FormatUint(c.gen(prefix), 10)
210
+
}
211
+
212
+
// keyPrefix recovers the object key / prefix a getter was asked for, stripping
213
+
// the "\x00window\x00gen" suffix groupKey/headKey append.
214
+
func keyPrefix(key string) string {
215
+
if i := strings.IndexByte(key, 0); i >= 0 {
216
+
return key[:i]
217
+
}
218
+
return key
219
+
}
220
+
221
+
// headInfo returns the object's HeadObject metadata, served from the head cache
222
+
// (prewarmed in the background from listings, or filled on demand here).
223
+
func (c *ListingCache) headInfo(ctx context.Context, objKey string) (*s3.HeadObjectOutput, error) {
224
+
var data []byte
225
+
if err := c.headGroup.Get(ctx, c.headKey(objKey), groupcache.AllocatingByteSliceSink(&data)); err != nil {
226
+
return nil, err
227
+
}
228
+
var d headData
229
+
if err := json.Unmarshal(data, &d); err != nil {
230
+
return nil, err
231
+
}
232
+
return &s3.HeadObjectOutput{
233
+
ContentLength: aws.Int64(d.Size),
234
+
LastModified: aws.Time(time.Unix(0, d.Mtime)),
235
+
Metadata: d.Meta,
236
+
}, nil
237
+
}
238
+
239
+
// prefetchHeads warms the head cache for every file in a freshly-listed folder,
240
+
// each via a detached background goroutine so the work never blocks the request
241
+
// that triggered the listing. Beyond maxPrefetchInFlight concurrent precaches it
242
+
// drops extras (logged), which are then fetched on demand.
243
+
func (c *ListingCache) prefetchHeads(prefix string, entries []childEntry) {
244
+
if c.cfg.DisableHeadPrefetch {
245
+
return
246
+
}
247
+
for _, e := range entries {
248
+
if e.Kind != kindFile {
249
+
continue
250
+
}
251
+
objKey := prefix + e.Name
252
+
select {
253
+
case c.prefetchSem <- struct{}{}:
254
+
default:
255
+
slog.Debug("head precache skipped: prefetch pipeline full", "key", objKey)
256
+
continue
257
+
}
258
+
go func(k string) {
259
+
defer func() { <-c.prefetchSem }()
260
+
var data []byte
261
+
if err := c.headGroup.Get(context.Background(), c.headKey(k), groupcache.AllocatingByteSliceSink(&data)); err != nil {
262
+
slog.Debug("head precache failed", "key", k, "err", err)
263
+
}
264
+
}(objKey)
265
+
}
266
+
}
267
+
268
+
// isNotFound reports whether err is an S3 "object does not exist" error.
269
+
func isNotFound(err error) bool {
270
+
var apiErr smithy.APIError
271
+
if errors.As(err, &apiErr) {
272
+
switch apiErr.ErrorCode() {
273
+
case "NotFound", "NoSuchKey":
274
+
return true
275
+
}
276
+
}
277
+
return false
278
+
}
279
+
280
+
func (c *ListingCache) gen(prefix string) uint64 {
281
+
c.mu.Lock()
282
+
defer c.mu.Unlock()
283
+
return c.gens[prefix]
284
+
}
285
+
286
+
// invalidate bumps a prefix's local generation so the next read in this process
287
+
// re-lists. groupcache cannot delete; moving the key is how we invalidate.
288
+
func (c *ListingCache) invalidate(prefix string) {
289
+
if c == nil {
290
+
return
291
+
}
292
+
c.mu.Lock()
293
+
c.gens[prefix]++
294
+
c.seen[prefix] = c.clock()
295
+
c.mu.Unlock()
296
+
}
297
+
298
+
func (c *ListingCache) touch(prefix string) {
299
+
c.mu.Lock()
300
+
c.seen[prefix] = c.clock()
301
+
c.mu.Unlock()
302
+
}
303
+
304
+
// RunWarmer pre-fills the current-window key for every recently-accessed prefix
305
+
// on each tick, smoothing window-rollover misses, and evicts prefixes unused for
306
+
// longer than IdleTTL. It returns when ctx is cancelled; it is a no-op when the
307
+
// refresh interval is non-positive.
308
+
func (c *ListingCache) RunWarmer(ctx context.Context) {
309
+
if c == nil || c.cfg.RefreshInterval <= 0 {
310
+
<-ctx.Done()
311
+
return
312
+
}
313
+
t := time.NewTicker(c.cfg.RefreshInterval)
314
+
defer t.Stop()
315
+
for {
316
+
select {
317
+
case <-ctx.Done():
318
+
return
319
+
case <-t.C:
320
+
c.warmOnce(ctx)
321
+
}
322
+
}
323
+
}
324
+
325
+
func (c *ListingCache) warmOnce(ctx context.Context) {
326
+
now := c.clock()
327
+
c.mu.Lock()
328
+
prefixes := make([]string, 0, len(c.seen))
329
+
for p, last := range c.seen {
330
+
if c.cfg.IdleTTL > 0 && now.Sub(last) > c.cfg.IdleTTL {
331
+
delete(c.seen, p)
332
+
continue
333
+
}
334
+
prefixes = append(prefixes, p)
335
+
}
336
+
c.mu.Unlock()
337
+
338
+
for _, p := range prefixes {
339
+
var data []byte
340
+
if err := c.group.Get(ctx, c.groupKey(p), groupcache.AllocatingByteSliceSink(&data)); err != nil {
341
+
slog.Debug("listing-cache warm failed", "prefix", p, "err", err)
342
+
}
343
+
}
344
+
}
345
+
346
+
// PoolHandler returns the groupcache peer HTTP handler, or nil in single-process
347
+
// mode. Serve it on the address passed as -groupcache-bind.
348
+
func (c *ListingCache) PoolHandler() http.Handler {
349
+
if c == nil || c.pool == nil {
350
+
return nil
351
+
}
352
+
return c.pool
353
+
}
354
+
355
+
// CacheStats is a flat snapshot of groupcache counters for export to metrics.
356
+
type CacheStats struct {
357
+
Gets, CacheHits, Loads, LocalLoads, PeerLoads, LocalLoadErrs int64
358
+
MainBytes, MainItems, MainEvictions int64
359
+
HotBytes, HotItems int64
360
+
}
361
+
362
+
// Stats snapshots the underlying groupcache counters.
363
+
func (c *ListingCache) Stats() CacheStats {
364
+
if c == nil || c.group == nil {
365
+
return CacheStats{}
366
+
}
367
+
main := c.group.CacheStats(groupcache.MainCache)
368
+
hot := c.group.CacheStats(groupcache.HotCache)
369
+
return CacheStats{
370
+
Gets: c.group.Stats.Gets.Get(),
371
+
CacheHits: c.group.Stats.CacheHits.Get(),
372
+
Loads: c.group.Stats.Loads.Get(),
373
+
LocalLoads: c.group.Stats.LocalLoads.Get(),
374
+
PeerLoads: c.group.Stats.PeerLoads.Get(),
375
+
LocalLoadErrs: c.group.Stats.LocalLoadErrs.Get(),
376
+
MainBytes: main.Bytes,
377
+
MainItems: main.Items,
378
+
MainEvictions: main.Evictions,
379
+
HotBytes: hot.Bytes,
380
+
HotItems: hot.Items,
381
+
}
382
+
}
383
+
384
+
// splitKey splits a canonical key into its parent prefix and base name. The
385
+
// prefix matches ReadDir's convention: "" for a root-level key, otherwise it
386
+
// ends in "/".
387
+
func splitKey(key string) (prefix, base string) {
388
+
if i := strings.LastIndex(key, "/"); i >= 0 {
389
+
return key[:i+1], key[i+1:]
390
+
}
391
+
return "", key
392
+
}
+383
internal/s3fs/listingcache_test.go
+383
internal/s3fs/listingcache_test.go
···
1
+
package s3fs
2
+
3
+
import (
4
+
"bytes"
5
+
"context"
6
+
"errors"
7
+
"fmt"
8
+
"io"
9
+
"io/fs"
10
+
"sort"
11
+
"strings"
12
+
"sync"
13
+
"sync/atomic"
14
+
"testing"
15
+
"time"
16
+
17
+
"github.com/aws/aws-sdk-go-v2/aws"
18
+
"github.com/aws/aws-sdk-go-v2/service/s3"
19
+
"github.com/aws/aws-sdk-go-v2/service/s3/types"
20
+
"github.com/aws/smithy-go"
21
+
"github.com/go-git/go-billy/v6"
22
+
)
23
+
24
+
// stubClient is a counting, in-memory s3Client for cache tests. It honours
25
+
// Prefix+Delimiter listing so ReadDir/Stat resolution behaves like S3.
26
+
type stubClient struct {
27
+
mu sync.Mutex
28
+
keys map[string]int64 // object key -> size
29
+
30
+
heads atomic.Int64
31
+
lists atomic.Int64
32
+
puts atomic.Int64
33
+
}
34
+
35
+
func newStub(keys ...string) *stubClient {
36
+
s := &stubClient{keys: map[string]int64{}}
37
+
for _, k := range keys {
38
+
s.keys[k] = 0
39
+
}
40
+
return s
41
+
}
42
+
43
+
func notFound() error { return &smithy.GenericAPIError{Code: "NotFound", Message: "not found"} }
44
+
45
+
func (s *stubClient) HeadObject(_ context.Context, in *s3.HeadObjectInput, _ ...func(*s3.Options)) (*s3.HeadObjectOutput, error) {
46
+
s.heads.Add(1)
47
+
s.mu.Lock()
48
+
size, ok := s.keys[aws.ToString(in.Key)]
49
+
s.mu.Unlock()
50
+
if !ok {
51
+
return nil, notFound()
52
+
}
53
+
return &s3.HeadObjectOutput{ContentLength: aws.Int64(size), LastModified: aws.Time(time.Unix(0, 0))}, nil
54
+
}
55
+
56
+
func (s *stubClient) ListObjectsV2(_ context.Context, in *s3.ListObjectsV2Input, _ ...func(*s3.Options)) (*s3.ListObjectsV2Output, error) {
57
+
s.lists.Add(1)
58
+
prefix := aws.ToString(in.Prefix)
59
+
delim := aws.ToString(in.Delimiter)
60
+
61
+
s.mu.Lock()
62
+
ks := make([]string, 0, len(s.keys))
63
+
for k := range s.keys {
64
+
ks = append(ks, k)
65
+
}
66
+
sizes := make(map[string]int64, len(s.keys))
67
+
for k, v := range s.keys {
68
+
sizes[k] = v
69
+
}
70
+
s.mu.Unlock()
71
+
sort.Strings(ks)
72
+
73
+
seenCP := map[string]bool{}
74
+
var cps []types.CommonPrefix
75
+
var contents []types.Object
76
+
for _, k := range ks {
77
+
if !strings.HasPrefix(k, prefix) {
78
+
continue
79
+
}
80
+
rest := k[len(prefix):]
81
+
if delim != "" {
82
+
if i := strings.Index(rest, delim); i >= 0 {
83
+
cp := prefix + rest[:i+1]
84
+
if !seenCP[cp] {
85
+
seenCP[cp] = true
86
+
cps = append(cps, types.CommonPrefix{Prefix: aws.String(cp)})
87
+
}
88
+
continue
89
+
}
90
+
}
91
+
contents = append(contents, types.Object{
92
+
Key: aws.String(k),
93
+
Size: aws.Int64(sizes[k]),
94
+
LastModified: aws.Time(time.Unix(0, 0)),
95
+
})
96
+
}
97
+
return &s3.ListObjectsV2Output{Contents: contents, CommonPrefixes: cps, IsTruncated: aws.Bool(false)}, nil
98
+
}
99
+
100
+
func (s *stubClient) PutObject(_ context.Context, in *s3.PutObjectInput, _ ...func(*s3.Options)) (*s3.PutObjectOutput, error) {
101
+
s.puts.Add(1)
102
+
s.mu.Lock()
103
+
s.keys[aws.ToString(in.Key)] = 0
104
+
s.mu.Unlock()
105
+
return &s3.PutObjectOutput{}, nil
106
+
}
107
+
108
+
func (s *stubClient) GetObject(_ context.Context, in *s3.GetObjectInput, _ ...func(*s3.Options)) (*s3.GetObjectOutput, error) {
109
+
return &s3.GetObjectOutput{Body: io.NopCloser(bytes.NewReader(nil))}, nil
110
+
}
111
+
112
+
func (s *stubClient) DeleteObject(_ context.Context, in *s3.DeleteObjectInput, _ ...func(*s3.Options)) (*s3.DeleteObjectOutput, error) {
113
+
s.mu.Lock()
114
+
delete(s.keys, aws.ToString(in.Key))
115
+
s.mu.Unlock()
116
+
return &s3.DeleteObjectOutput{}, nil
117
+
}
118
+
119
+
func (s *stubClient) RenameObject(_ context.Context, in *s3.CopyObjectInput, _ ...func(*s3.Options)) (*s3.CopyObjectOutput, error) {
120
+
return &s3.CopyObjectOutput{}, nil
121
+
}
122
+
123
+
func (s *stubClient) CreateMultipartUpload(_ context.Context, in *s3.CreateMultipartUploadInput, _ ...func(*s3.Options)) (*s3.CreateMultipartUploadOutput, error) {
124
+
return &s3.CreateMultipartUploadOutput{UploadId: aws.String("u")}, nil
125
+
}
126
+
127
+
func (s *stubClient) UploadPart(_ context.Context, in *s3.UploadPartInput, _ ...func(*s3.Options)) (*s3.UploadPartOutput, error) {
128
+
return &s3.UploadPartOutput{}, nil
129
+
}
130
+
131
+
func (s *stubClient) CompleteMultipartUpload(_ context.Context, in *s3.CompleteMultipartUploadInput, _ ...func(*s3.Options)) (*s3.CompleteMultipartUploadOutput, error) {
132
+
return &s3.CompleteMultipartUploadOutput{}, nil
133
+
}
134
+
135
+
// gcSeq gives every test cache a unique groupcache group name; NewGroup panics
136
+
// on a duplicate name.
137
+
var gcSeq atomic.Int64
138
+
139
+
// newTestCache disables the background head precache so S3-call counts are
140
+
// deterministic; TestListingCacheHeadPrefetch exercises it explicitly.
141
+
func newTestCache(stub *stubClient, ttl time.Duration) *ListingCache {
142
+
n := gcSeq.Add(1)
143
+
return NewListingCache(CacheConfig{TTL: ttl, Name: fmt.Sprintf("test-%d", n), DisableHeadPrefetch: true}, stub, "bucket", "/")
144
+
}
145
+
146
+
func newCachedFS(t *testing.T, stub *stubClient, cache *ListingCache) billy.Filesystem {
147
+
t.Helper()
148
+
fsys, err := NewS3FS(stub, "bucket", WithListingCache(cache))
149
+
if err != nil {
150
+
t.Fatalf("NewS3FS: %v", err)
151
+
}
152
+
return fsys
153
+
}
154
+
155
+
func TestListingCachePopulateOnMiss(t *testing.T) {
156
+
stub := newStub("objects/ab/file1")
157
+
cache := newTestCache(stub, time.Hour)
158
+
fsys := newCachedFS(t, stub, cache)
159
+
160
+
// First Stat of an absent key lists the parent once, no HeadObject.
161
+
if _, err := fsys.Stat("objects/ab/nope"); !errors.Is(err, fs.ErrNotExist) {
162
+
t.Fatalf("Stat absent: want ErrNotExist, got %v", err)
163
+
}
164
+
if got := stub.lists.Load(); got != 1 {
165
+
t.Fatalf("first miss: lists = %d, want 1", got)
166
+
}
167
+
if got := stub.heads.Load(); got != 0 {
168
+
t.Fatalf("first miss: heads = %d, want 0", got)
169
+
}
170
+
171
+
// A second absent sibling is a pure negative hit: no S3 at all.
172
+
l0, h0 := stub.lists.Load(), stub.heads.Load()
173
+
if _, err := fsys.Stat("objects/ab/nope2"); !errors.Is(err, fs.ErrNotExist) {
174
+
t.Fatalf("Stat absent2: want ErrNotExist, got %v", err)
175
+
}
176
+
if stub.lists.Load() != l0 || stub.heads.Load() != h0 {
177
+
t.Fatalf("negative hit did S3: lists %d->%d heads %d->%d", l0, stub.lists.Load(), h0, stub.heads.Load())
178
+
}
179
+
180
+
// A present file resolves from cache but still HeadObjects for metadata.
181
+
l0, h0 = stub.lists.Load(), stub.heads.Load()
182
+
fi, err := fsys.Stat("objects/ab/file1")
183
+
if err != nil {
184
+
t.Fatalf("Stat present: %v", err)
185
+
}
186
+
if fi.IsDir() {
187
+
t.Fatalf("Stat present: reported a directory")
188
+
}
189
+
if stub.lists.Load() != l0 {
190
+
t.Fatalf("present file re-listed: %d->%d", l0, stub.lists.Load())
191
+
}
192
+
if stub.heads.Load() != h0+1 {
193
+
t.Fatalf("present file heads = %d, want %d", stub.heads.Load(), h0+1)
194
+
}
195
+
}
196
+
197
+
func TestListingCacheReadDirThenStat(t *testing.T) {
198
+
stub := newStub("objects/ab/file1", "objects/ab/sub/deep")
199
+
cache := newTestCache(stub, time.Hour)
200
+
fsys := newCachedFS(t, stub, cache)
201
+
202
+
entries, err := fsys.ReadDir("objects/ab")
203
+
if err != nil {
204
+
t.Fatalf("ReadDir: %v", err)
205
+
}
206
+
if len(entries) != 2 {
207
+
t.Fatalf("ReadDir entries = %d, want 2", len(entries))
208
+
}
209
+
if stub.lists.Load() != 1 {
210
+
t.Fatalf("ReadDir lists = %d, want 1", stub.lists.Load())
211
+
}
212
+
213
+
// The sub-directory resolves from the cached listing with no S3 calls.
214
+
l0, h0 := stub.lists.Load(), stub.heads.Load()
215
+
fi, err := fsys.Stat("objects/ab/sub")
216
+
if err != nil {
217
+
t.Fatalf("Stat dir: %v", err)
218
+
}
219
+
if !fi.IsDir() {
220
+
t.Fatalf("Stat dir: not a directory")
221
+
}
222
+
if stub.lists.Load() != l0 || stub.heads.Load() != h0 {
223
+
t.Fatalf("dir resolve did S3: lists %d->%d heads %d->%d", l0, stub.lists.Load(), h0, stub.heads.Load())
224
+
}
225
+
226
+
// And an absent sibling is a negative hit.
227
+
if _, err := fsys.Stat("objects/ab/missing"); !errors.Is(err, fs.ErrNotExist) {
228
+
t.Fatalf("Stat missing: want ErrNotExist, got %v", err)
229
+
}
230
+
if stub.lists.Load() != l0 || stub.heads.Load() != h0 {
231
+
t.Fatalf("negative hit did S3 after ReadDir")
232
+
}
233
+
}
234
+
235
+
func TestListingCacheInvalidatesOnWrite(t *testing.T) {
236
+
stub := newStub("refs/heads/main")
237
+
cache := newTestCache(stub, time.Hour)
238
+
fsys := newCachedFS(t, stub, cache)
239
+
240
+
// Warm the cache and confirm the new ref reads as absent.
241
+
if _, err := fsys.Stat("refs/heads/new"); !errors.Is(err, fs.ErrNotExist) {
242
+
t.Fatalf("Stat before write: want ErrNotExist, got %v", err)
243
+
}
244
+
245
+
// Write the ref; Close must invalidate the parent prefix.
246
+
f, err := fsys.Create("refs/heads/new")
247
+
if err != nil {
248
+
t.Fatalf("Create: %v", err)
249
+
}
250
+
if _, err := f.Write([]byte("deadbeef")); err != nil {
251
+
t.Fatalf("Write: %v", err)
252
+
}
253
+
if err := f.Close(); err != nil {
254
+
t.Fatalf("Close: %v", err)
255
+
}
256
+
257
+
// The next Stat must re-list (generation bumped) and now see the ref.
258
+
fi, err := fsys.Stat("refs/heads/new")
259
+
if err != nil {
260
+
t.Fatalf("Stat after write: %v", err)
261
+
}
262
+
if fi.IsDir() {
263
+
t.Fatalf("Stat after write: reported a directory")
264
+
}
265
+
}
266
+
267
+
func TestListingCacheWindowExpiry(t *testing.T) {
268
+
stub := newStub("objects/ab/file1")
269
+
cache := newTestCache(stub, time.Minute)
270
+
now := time.Unix(1_000_000, 0)
271
+
cache.clock = func() time.Time { return now }
272
+
fsys := newCachedFS(t, stub, cache)
273
+
274
+
if _, err := fsys.Stat("objects/ab/nope"); !errors.Is(err, fs.ErrNotExist) {
275
+
t.Fatal("Stat 1")
276
+
}
277
+
if stub.lists.Load() != 1 {
278
+
t.Fatalf("lists = %d, want 1", stub.lists.Load())
279
+
}
280
+
// Same window: served from cache.
281
+
if _, err := fsys.Stat("objects/ab/nope"); !errors.Is(err, fs.ErrNotExist) {
282
+
t.Fatal("Stat 2")
283
+
}
284
+
if stub.lists.Load() != 1 {
285
+
t.Fatalf("same window re-listed: lists = %d, want 1", stub.lists.Load())
286
+
}
287
+
// Advance past the TTL window: the key changes and we re-list.
288
+
now = now.Add(2 * time.Minute)
289
+
if _, err := fsys.Stat("objects/ab/nope"); !errors.Is(err, fs.ErrNotExist) {
290
+
t.Fatal("Stat 3")
291
+
}
292
+
if stub.lists.Load() != 2 {
293
+
t.Fatalf("new window lists = %d, want 2", stub.lists.Load())
294
+
}
295
+
}
296
+
297
+
func TestListingCacheWarmer(t *testing.T) {
298
+
stub := newStub("objects/ab/file1")
299
+
cache := newTestCache(stub, time.Hour)
300
+
cache.cfg.RefreshInterval = time.Millisecond // enable warming
301
+
cache.cfg.IdleTTL = time.Minute
302
+
now := time.Unix(1_000_000, 0)
303
+
cache.clock = func() time.Time { return now }
304
+
305
+
// Touch a prefix, then warm: the warmer fills it (one list).
306
+
cache.touch("objects/ab/")
307
+
cache.warmOnce(context.Background())
308
+
if stub.lists.Load() != 1 {
309
+
t.Fatalf("warmer lists = %d, want 1", stub.lists.Load())
310
+
}
311
+
312
+
// After the idle window the warmer evicts the prefix from its working set.
313
+
now = now.Add(2 * time.Minute)
314
+
cache.warmOnce(context.Background())
315
+
cache.mu.Lock()
316
+
n := len(cache.seen)
317
+
cache.mu.Unlock()
318
+
if n != 0 {
319
+
t.Fatalf("idle prefix not evicted: seen = %d", n)
320
+
}
321
+
}
322
+
323
+
func TestListingCacheHeadPrefetch(t *testing.T) {
324
+
stub := newStub("objects/ab/f1", "objects/ab/f2")
325
+
n := gcSeq.Add(1)
326
+
cache := NewListingCache(CacheConfig{TTL: time.Hour, Name: fmt.Sprintf("test-%d", n)}, stub, "bucket", "/")
327
+
fsys := newCachedFS(t, stub, cache)
328
+
329
+
// A single listing fill (here triggered by an absent lookup) prefetches the
330
+
// HeadObject for every file in the folder, in the background.
331
+
if _, err := fsys.Stat("objects/ab/nope"); !errors.Is(err, fs.ErrNotExist) {
332
+
t.Fatalf("Stat: %v", err)
333
+
}
334
+
335
+
// Wait for the two background HeadObjects to land.
336
+
deadline := time.Now().Add(2 * time.Second)
337
+
for stub.heads.Load() < 2 {
338
+
if time.Now().After(deadline) {
339
+
t.Fatalf("prefetch did not warm both heads: got %d", stub.heads.Load())
340
+
}
341
+
time.Sleep(time.Millisecond)
342
+
}
343
+
if got := stub.heads.Load(); got != 2 {
344
+
t.Fatalf("prefetch heads = %d, want 2", got)
345
+
}
346
+
347
+
// Stat of a prewarmed file is served from the head cache: no new HeadObject.
348
+
h0 := stub.heads.Load()
349
+
if _, err := fsys.Stat("objects/ab/f1"); err != nil {
350
+
t.Fatalf("Stat f1: %v", err)
351
+
}
352
+
if stub.heads.Load() != h0 {
353
+
t.Fatalf("prewarmed Stat did a HeadObject: %d->%d", h0, stub.heads.Load())
354
+
}
355
+
}
356
+
357
+
func TestListingCacheChrootShares(t *testing.T) {
358
+
stub := newStub("repo/objects/ab/file1")
359
+
cache := newTestCache(stub, time.Hour)
360
+
rootfs := newCachedFS(t, stub, cache)
361
+
362
+
// Populate the listing for repo/objects/ab/ via the root view.
363
+
if _, err := rootfs.Stat("repo/objects/ab/missing1"); !errors.Is(err, fs.ErrNotExist) {
364
+
t.Fatalf("root Stat: want ErrNotExist, got %v", err)
365
+
}
366
+
if stub.lists.Load() != 1 {
367
+
t.Fatalf("root Stat lists = %d, want 1", stub.lists.Load())
368
+
}
369
+
370
+
// A chroot view shares the same cache keyed by canonical prefix, so a Stat
371
+
// under it hits the cached listing with no further S3.
372
+
sub, err := rootfs.Chroot("repo")
373
+
if err != nil {
374
+
t.Fatalf("Chroot: %v", err)
375
+
}
376
+
l0 := stub.lists.Load()
377
+
if _, err := sub.Stat("objects/ab/missing2"); !errors.Is(err, fs.ErrNotExist) {
378
+
t.Fatalf("chroot Stat: want ErrNotExist, got %v", err)
379
+
}
380
+
if stub.lists.Load() != l0 {
381
+
t.Fatalf("chroot Stat re-listed: %d->%d", l0, stub.lists.Load())
382
+
}
383
+
}