diff --git a/cmd/objgitd/main.go b/cmd/objgitd/main.go index 031ba5a..65bf0e2 100644 --- a/cmd/objgitd/main.go +++ b/cmd/objgitd/main.go @@ -12,6 +12,7 @@ import ( "os" "os/signal" "runtime" + "strings" "syscall" "time" @@ -40,6 +41,15 @@ var ( allowHooks = flag.Bool("allow-hooks", false, "run .objgit/hooks/receive-pack in a sandbox after a successful push") hookTimeout = flag.Duration("hook-timeout", 60*time.Second, "wall-clock limit for a single hook run") + + 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") + s3CacheRefresh = flag.Duration("s3-cache-refresh", 30*time.Second, "interval at which the listing-cache warmer re-fills hot prefixes; 0 disables the warmer") + s3CacheIdle = flag.Duration("s3-cache-idle", 10*time.Minute, "drop a prefix from the listing-cache warmer after this long without access") + s3CacheSize = flag.Int64("s3-cache-size", 64<<20, "groupcache LRU budget in bytes for the listing cache") + + 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") + groupcachePeers = flag.String("groupcache-peers", "", "comma-separated groupcache peer base URLs (including self) for cross-process sharing") + groupcacheBind = flag.String("groupcache-bind", "", "TCP address to serve the groupcache peer endpoint; empty disables peer sharing") ) func main() { @@ -75,7 +85,34 @@ func main() { os.Exit(1) } - fsys, err := s3fs.NewS3FS(client, *bucket) + var cache *s3fs.ListingCache + var fsOpts []s3fs.Option + if *s3CacheTTL > 0 { + var peers []string + if *groupcachePeers != "" { + peers = strings.Split(*groupcachePeers, ",") + } + cache = s3fs.NewListingCache(s3fs.CacheConfig{ + TTL: *s3CacheTTL, + RefreshInterval: *s3CacheRefresh, + IdleTTL: *s3CacheIdle, + SizeBytes: *s3CacheSize, + Self: *groupcacheSelf, + Peers: peers, + }, client, *bucket, "/") + fsOpts = append(fsOpts, s3fs.WithListingCache(cache)) + metrics.RegisterListingCache(func() metrics.ListingCacheStats { + s := cache.Stats() + return metrics.ListingCacheStats{ + Gets: s.Gets, CacheHits: s.CacheHits, Loads: s.Loads, + LocalLoads: s.LocalLoads, PeerLoads: s.PeerLoads, LocalLoadErrs: s.LocalLoadErrs, + MainBytes: s.MainBytes, MainItems: s.MainItems, MainEvictions: s.MainEvictions, + HotBytes: s.HotBytes, HotItems: s.HotItems, + } + }) + } + + fsys, err := s3fs.NewS3FS(client, *bucket, fsOpts...) if err != nil { slog.Error("can't create s3fs", "bucket", *bucket, "err", err) os.Exit(1) @@ -97,10 +134,46 @@ func main() { "bucket", *bucket, "allow_push", *allowPush, "allow_hooks", *allowHooks, + "s3_cache_ttl", *s3CacheTTL, + "s3_cache_refresh", *s3CacheRefresh, + "groupcache_self", *groupcacheSelf, ) g, gCtx := errgroup.WithContext(ctx) + if cache != nil { + g.Go(func() error { + cache.RunWarmer(gCtx) + return nil + }) + + if *groupcacheBind != "" { + handler := cache.PoolHandler() + if handler == nil { + slog.Error("-groupcache-bind set without -groupcache-self; peer sharing is disabled") + os.Exit(1) + } + ln, err := net.Listen("tcp", *groupcacheBind) + if err != nil { + slog.Error("can't listen", "groupcache_bind", *groupcacheBind, "err", err) + os.Exit(1) + } + srv := &http.Server{Handler: handler} + g.Go(func() error { + if err := srv.Serve(ln); err != nil && !errors.Is(err, http.ErrServerClosed) { + return err + } + return nil + }) + g.Go(func() error { + <-gCtx.Done() + shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + return srv.Shutdown(shutdownCtx) + }) + } + } + if *metricsBind != "" { ln, err := net.Listen("tcp", *metricsBind) if err != nil { diff --git a/docs/plans/s3fs-groupcache-listobjectsv2.md b/docs/plans/s3fs-groupcache-listobjectsv2.md new file mode 100644 index 0000000..7d37aaa --- /dev/null +++ b/docs/plans/s3fs-groupcache-listobjectsv2.md @@ -0,0 +1,267 @@ +# Plan: groupcache-backed directory-listing cache for `internal/s3fs` + +## Context + +`objgitd` stores git repos as S3/Tigris objects. Every `Stat`/`Open` on a path that +**doesn't exist** costs up to **two** S3 round-trips: a `HeadObject` (→ `NoSuchKey`) +then a directory-probe `ListObjectsV2` (`internal/s3fs/basic.go:140`+`:163`; the probe +in `OpenFile` at `:98`). git does an enormous number of these — loose objects on a +packed repo, `packed-refs` vs. loose refs, `info/`, `config`, alternates, `.keep`/`.idx` +siblings — each paid in series against an object store. + +The fix: cache **directory listings**, keyed by parent prefix, and answer `Stat`/`Open` +from them. A listing records every child name plus its kind/size/mtime, so a lookup of a +key whose parent prefix is cached answers with **zero** round-trips when the child is +absent (negative hit) or is a sub-directory, and with one `HeadObject`/`GetObject` (for +authoritative content/metadata, which listings don't carry) when it's a file. The first +`Stat`/`Open` touching an un-cached folder lists that **parent folder** in full and +populates the cache, warming every sibling at once. + +**Backend (user decision): `github.com/golang/groupcache`,** so the listing cache is +shared across processes via groupcache's consistent-hash peer pool. groupcache is a +fill-once, **immutable** cache: no `Set`, no `Delete`, no TTL. We adapt to that with two +techniques baked into the cache **key**: + +- **Window-encoded TTL.** The key carries a time window `floor(now/TTL)`. When the + window advances the key changes → groupcache miss → the getter re-lists from S3. Old + keys age out by LRU. Staleness is bounded by one TTL window. +- **Per-prefix local generation for precise in-process invalidation.** The key also + carries a process-local counter per prefix, bumped on every local write under that + prefix. A local write moves the key, so the next read in this process re-lists and + sees its own write immediately — recovering read-after-write correctness that + groupcache alone cannot give (no delete). The stale entry under the old key is simply + never queried again by us. + +**Accepted limitation (explicit user choice):** cross-process and post-restart staleness +is bounded by the TTL window — another process (or this one after a restart that resets +the local generation) may read a just-written object as "not found" for up to one +window. This is safe for git's _content_ (objects are immutable and content-addressed, so +positive listings never go wrong), and the window bounds the negative-staleness risk. +Operators tune it with `-s3-cache-ttl`; `-s3-cache-ttl 0` disables the cache and restores +today's exact behavior. + +**Defaults:** on by default, tunable; single-process unless peers are configured. + +## Correctness model + +- **Cache key space is the canonical full S3 key.** `Chroot` returns a _new_ `*S3FS` + with a different `root`, so the `*ListingCache` is **shared by pointer** across a root + fs and all chroot children, and prefixes are full-canonical (`fs3.key()`/`cleanPath` — + root-joined, leading slash stripped). `chroot.go:21` must copy the cache pointer. +- **groupcache key** = `prefix "\x00" window "\x00" localgen`, where + `window = now.Unix() / int64(ttl.Seconds())` and `localgen = c.gen(prefix)`. The + getter parses `prefix` back off the key (segment before the first `\x00`) and lists it. +- **Population is via the getter only** (groupcache has no `Set`). Both `ReadDir` and the + `Stat`/`Open` resolution route through `group.Get`, so each folder is listed once per + (window, localgen) and the result feeds both. groupcache's singleflight dedupes + concurrent identical `Get`s (one list even when a clone fans out across siblings), and + the owning peer dedupes across the fleet. +- **Listing payload** carries enough to serve both consumers without a second call: + per child `{name, kind(file|dir), size, mtimeUnixNano}` (`CommonPrefixes`→dir with + zero size/mtime, `Contents`→file; file wins on the pathological file+prefix collision, + matching today's Head-first precedence). Encoded compactly (JSON is fine; listings are + small). +- **In-process writes bump the local generation** of the parent prefix at _write + completion_, moving the key so the next read re-lists: + - `s3WriteFile.Close` / `s3MultipartUploadFile.Close` — after the upload succeeds. + - `Rename` (covers `TempFile`→final pack promotion), `Remove`, `MkdirAll`. +- **Positive `Stat`/`Open` of a file** is served from a second cache, the **head + cache** (see [Head precache](#head-precache)), rather than a foreground `HeadObject`. + Listings omit the user-metadata the unix-metadata feature needs, so head metadata is + cached per object and **prefetched in the background** when a listing is filled. `Open` + still issues `GetObject` for the body but skips its `HeadObject`. A `NoSuchKey` from a + delete racing the listing maps to `NotExist`. + +## Changes + +### go.mod + +- Add `github.com/golang/groupcache` (effectively dependency-free). + +### New file: `internal/s3fs/listingcache.go` + +The cache wrapper, groupcache group, key logic, and background warmer. + +```go +type CacheConfig struct { + TTL, RefreshInterval, IdleTTL time.Duration + SizeBytes int64 // groupcache LRU budget + Name string // group name (default "objgit-listings") + DisableHeadPrefetch bool // zero value = background head prefetch on + Self string // this node's groupcache URL ("" = single-process) + Peers []string // peer URLs (incl. self) when sharing +} + +type childKind uint8 // kindFile / kindDir +type childEntry struct { Name string; Kind childKind; Size, Mtime int64 } +type headData struct { Size, Mtime int64; Meta map[string]string } + +type ListingCache struct { + group *groupcache.Group // listings, keyed by prefix + headGroup *groupcache.Group // HeadObject metadata, keyed by object key + pool *groupcache.HTTPPool // nil in single-process mode + ttl time.Duration + cfg CacheConfig + clock func() time.Time // overridable in tests + prefetchSem chan struct{} // bounds background head precaches + mu sync.Mutex + gens map[string]uint64 // per-prefix local generation + seen map[string]time.Time // prefixes accessed → driven by the warmer +} +``` + +- `NewListingCache(cfg, client, bucket, separator) *ListingCache` — creates the + groupcache `Group` (name `"objgit-listings"`, `cfg.SizeBytes`) with a `GetterFunc` + closure over `client`/`bucket`/`separator` that parses the prefix from the key, runs a + full paginated `listChildren`, and encodes the payload into the sink. When + `cfg.Self != ""`, builds a `groupcache.NewHTTPPoolOpts(cfg.Self, …)` and `pool.Set(cfg.Peers…)`. +- `list(ctx, prefix) ([]childEntry, error)` — record `seen[prefix]=now`; build the + groupcache key; `group.Get(ctx, key, AllocatingByteSliceSink(&buf))`; decode. This is + the one entry point both `ReadDir` and `Stat`/`Open` use. +- `gen(prefix)` / `invalidate(prefix)` — `invalidate` bumps `gens[prefix]` and records + `seen` so the warmer re-warms the new key (no groupcache delete needed). +- `runWarmer(ctx)` / `RunWarmer(ctx)` — `time.NewTicker(RefreshInterval)`; each tick drop + `seen` entries idle past `IdleTTL`, then `group.Get` the current key for each remaining + prefix (pre-fills the new window before clients wait and smooths window-rollover + herds). No-op when `RefreshInterval<=0`. Returns on `ctx.Done()` (errgroup idiom). +- `PoolHandler() http.Handler` / `Stats()` accessors for `main` to serve peers and export + metrics. +- `splitKey(key) (prefix, base)` — split on last `/`; no slash → `("", key)`. + +### Head precache + +A **second groupcache group** (`-heads`, keyed by object key) caches each file's +`HeadObject` result — `headData{size, mtimeUnixNano, meta}` — so positive `Stat`/`Open` +never pay a foreground `HeadObject`. The head key reuses the window + the **parent +prefix's** local generation, so a write under that prefix re-heads the object exactly as +it re-lists the folder. + +- **Background prefetch (`prefetchHeads`).** When the listing getter fills a folder, it + launches a **detached goroutine per file** (`go headGroup.Get(context.Background(), …)`) + to warm the head cache off the request's critical path; groupcache singleflight + coalesces a background precache with any foreground head lookup for the same object. A + semaphore (`maxPrefetchInFlight = 64`) bounds the in-flight precaches so a large folder + can't spawn an unbounded goroutine/S3 storm — overflow files are fetched on demand + (logged at debug). `CacheConfig.DisableHeadPrefetch` turns this off (zero value = on). +- **`headInfo(ctx, key) (*s3.HeadObjectOutput, error)`** serves a foreground lookup from + the head cache (a warm hit costs no round-trip; a miss fills via one `HeadObject`). +- `Stat`/`Open` of a present file call `headInfo`; `newS3ReadFile` gains an optional + precomputed `*s3.HeadObjectOutput` so it skips its own `HeadObject` and only `GetObject`s + the body. + +### `internal/s3fs/filesystem.go` + +- Add `cache *ListingCache` to `S3FS`; `WithListingCache(c) Option`. `NewS3FS`'s `client` + parameter widens to an `s3Client` interface (the concrete `*storage.Client` satisfies + it) so tests can substitute a counting stub. + +### `internal/s3fs/chroot.go` + +- Copy `cache: fs3.cache` into the new `S3FS` literal (`chroot.go:21`). + +### `internal/s3fs/dir.go` + +- Extract `listChildren(ctx, client, bucket, separator, prefix) ([]childEntry, error)` — + the paginated `ListObjectsV2` loop, classifying `CommonPrefixes`→dir, `Contents`→file + with size/mtime, dirs-then-files preserving S3 order. A free function so the getter + (which holds only the raw client) can reuse it. Used by `ReadDir` (cache off) and both + getters. +- `ReadDir`: when `cache != nil`, get the entries via `cache.list(ctx, prefix)` and + build `[]fs.DirEntry` from them (rebuilding `newDirInfo`/`newFileInfo` from the payload); + otherwise list directly as today. +- `MkdirAll` — `cache.invalidate(parent prefix of filename)` after the `PutObject`. + +### `internal/s3fs/basic.go` + +- Helper `resolve(ctx, key) (childEntry, found, known bool)`: `(_,_,false)` when + `cache==nil`; else `prefix,base := splitKey(key)`, `entries, err := cache.list(...)`; on + error `known=false` (fall back to the live path — cache problems never fail the op); + else scan `entries` for `base` and return it. +- `Stat`: after the temp-buffer check, call `resolve`. If `known`: absent → + `&os.PathError{Op:"stat",…,Err: fs.ErrNotExist}`; dir → `newDirInfo`; file → + `cache.headInfo`→`newFileInfoFromHead` (head `NoSuchKey` → `NotExist`). Not `known` → the + existing `HeadObject`+probe fallback. +- `OpenFile` `O_RDONLY` (after temp check): same `resolve`; absent → `NotExist`, dir → + `newS3DirFile`, file → `cache.headInfo` then `newS3ReadFile(…, ho)` (skips its + `HeadObject`); not `known` → existing fallback with a nil head. +- `Rename` — invalidate parent prefix of both `src` and `dst` on success. +- `Remove` — invalidate parent prefix of `key` on success. + +### `internal/s3fs/file.go` + +- Add a `cache *ListingCache` field + constructor arg to `s3WriteFile` / + `s3MultipartUploadFile`; their `Close` calls `cache.invalidate(parent prefix of f.key)` + after a successful upload (nil-guarded). Update the two `OpenFile` call sites + (`basic.go:114`,`:117`). + +### `internal/metrics/metrics.go` + +- A Prometheus collector that reads `ListingCache.Stats()` (groupcache `group.Stats`: + Gets, CacheHits, Loads, LocalLoads, PeerLoads, LoaderErrors; plus `CacheStats` bytes/ + items/evictions) and exports them as `objgit_s3_listing_cache_*` gauges/counters. No + `repo` label. (groupcache fills are already counted as `ListObjectsV2` via `observeS3`.) + `main` registers it only when the cache is enabled. + +### `cmd/objgitd/main.go` + +- New flags (kebab-case + flagenv) in the `var (...)` block at `main.go:32`: + - `-s3-cache-ttl` (`Duration`, default `60s`) — window size; `<=0` disables the cache. + - `-s3-cache-refresh` (`Duration`, default `30s`) — warmer interval; `<=0` disables the + warmer (lazy fill still works). + - `-s3-cache-idle` (`Duration`, default `10m`) — drop un-accessed prefixes from the warmer. + - `-s3-cache-size` (`Int64`, default `64<<20`) — groupcache LRU budget. + - `-groupcache-self` (`String`, default `""`) — this node's groupcache base URL; empty = + single-process. + - `-groupcache-peers` (`String`, default `""`) — comma-separated peer URLs. + - `-groupcache-bind` (`String`, default `""`) — listen addr serving `PoolHandler()`. +- When `ttl > 0`: build `cache := s3fs.NewListingCache(cfg, client, *bucket, "/")`, pass + `s3fs.WithListingCache(cache)` into `NewS3FS` (`main.go:78`), register the metrics + collector, and in the errgroup: + - if `-groupcache-bind != ""`, `net.Listen` + serve `cache.PoolHandler()` (plus the + Serve/`Shutdown`-on-`gCtx.Done()` two-goroutine idiom used by the other listeners). + - `g.Go(func() error { cache.RunWarmer(gCtx); return nil })`. +- Add the cache settings to the startup `slog.Info` line. + +## Why this is acceptably safe + +Within a single git operation the repeated negative lookups happen within milliseconds, +so even a 60s window eliminates essentially all redundant `HeadObject`+probe pairs, and +the first miss in a folder warms every sibling via one parent listing (deduped by +groupcache singleflight, fleet-wide via the owning peer). Local writes bump the +per-prefix generation, so a push reads its own objects immediately. git object _content_ +is immutable, so positive/`ReadDir` results are never wrong about what exists — only the +_recency_ of newly-added entries is window-bounded. The residual risk is a cross-process +(or post-restart) negative read of an object another writer just created, bounded by one +TTL window — the limitation the user explicitly accepted; `-s3-cache-ttl 0` opts out +entirely. Note: writes fragment the shared keyspace for written prefixes (each writer's +`localgen` differs), so cross-process sharing is strongest on the read-only prefixes that +dominate clone/fetch and weakest on actively-pushed prefixes — which is the desirable +bias. + +## Verification + +1. `go build ./...`; `go mod tidy`; `go test ./...`. +2. **Cache disabled = no behavior change:** the cache is only wired when `-s3-cache-ttl>0`; + existing protocol tests (`go test ./cmd/objgitd/...`, needs `git` on PATH) must pass + with the cache off and on (single-process, no peers). +3. **New unit tests in `internal/s3fs`** (table-driven `tt`, counting-stub `storage.Client`): + - **Populate-on-miss:** first `Stat` of an absent key in a never-listed folder issues + exactly one `ListObjectsV2` (the parent) and **zero** `HeadObject`; a second absent + sibling → **zero** S3; a present sibling → only `HeadObject`. + - `ReadDir` then `Stat`/`Open` of an absent sibling → zero S3; a dir child → zero S3. + - **Local invalidation:** `Create`+`Close` / `Rename` / `Remove` / `MkdirAll` bump the + generation so a following `Stat`/`ReadDir` re-lists and sees the change (read-after-write). + - **Window TTL:** advancing time past `TTL` changes the key → re-list (inject a clock or + a settable `now` in `ListingCache` for the test). + - **Warmer:** `RunWarmer` re-`Get`s accessed prefixes and evicts idle ones past `IdleTTL`. + - **Chroot sharing:** `ReadDir` on the root then `Stat` of an absent child through a + chroot resolves from the same cached prefix (same canonical key). + - **Head precache:** one listing fill warms every file's head in the background; once + warm, `Stat` of a file does **zero** further `HeadObject`s (counting-stub tests + disable prefetch for determinism; a dedicated test enables it and waits for warmup). +4. **Two-process sharing (manual):** run two `objgitd` with `-groupcache-self`/`-peers` + pointing at each other; clone through one and confirm via `objgit_s3_listing_cache_*` + (PeerLoads > 0 on the non-owner) that listings are served cross-process. +5. **End-to-end:** `./objgitd -bucket $BUCKET -allow-push`; clone a packed repo twice and + confirm `objgit_s3_requests_total{operation="HeadObject"}` grows far slower than with + `-s3-cache-ttl 0`. diff --git a/go.mod b/go.mod index 4f0ad37..ab43cd4 100644 --- a/go.mod +++ b/go.mod @@ -10,6 +10,7 @@ require ( github.com/gliderlabs/ssh v0.3.8 github.com/go-git/go-billy/v6 v6.0.0-alpha.1 github.com/go-git/go-git/v6 v6.0.0-alpha.4 + github.com/golang/groupcache v0.0.0-20241129210726-2c02b8208cf8 github.com/joho/godotenv v1.5.1 github.com/prometheus/client_golang v1.23.2 github.com/tigrisdata/storage-go v0.6.0 @@ -47,6 +48,7 @@ require ( github.com/facebookgo/stack v0.0.0-20160209184415-751773369052 // indirect github.com/facebookgo/subset v0.0.0-20200203212716-c811ad88dec4 // indirect github.com/go-git/gcfg/v2 v2.0.2 // indirect + github.com/golang/protobuf v1.5.4 // indirect github.com/kevinburke/ssh_config v1.6.0 // indirect github.com/klauspost/cpuid/v2 v2.3.0 // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect diff --git a/go.sum b/go.sum index 482ad98..90f2ff3 100644 --- a/go.sum +++ b/go.sum @@ -75,6 +75,10 @@ github.com/go-git/go-git/v6 v6.0.0-alpha.4 h1:aDTc2UGanmaE7FkGLSlBEB9nohMnQ+RKXc github.com/go-git/go-git/v6 v6.0.0-alpha.4/go.mod h1:4ODa/G7hPWrh4Y+7lmt59Ij3zW38IEfvRoAZxLYYBhc= github.com/go-quicktest/qt v1.101.0 h1:O1K29Txy5P2OK0dGo59b7b0LR6wKfIhttaAhHUyn7eI= github.com/go-quicktest/qt v1.101.0/go.mod h1:14Bz/f7NwaXPtdYEgzsx46kqSxVwTbzVZsDC26tQJow= +github.com/golang/groupcache v0.0.0-20241129210726-2c02b8208cf8 h1:f+oWsMOmNPc8JmEHVZIycC7hBoQxHH9pNKQORJNozsQ= +github.com/golang/groupcache v0.0.0-20241129210726-2c02b8208cf8/go.mod h1:wcDNUvekVysuuOpQKo3191zZyTpiI6se1N1ULghS0sw= +github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= +github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0= diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index f55c0ac..26b2bf7 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -155,3 +155,63 @@ func ObserveHook(status string, dur time.Duration) { func ReposCreated() { reposCreated.Inc() } + +// ListingCacheStats is a flat snapshot of the s3fs directory-listing cache's +// groupcache counters. It mirrors s3fs.CacheStats but is defined here so s3fs +// stays free of any Prometheus import; main bridges the two. +type ListingCacheStats struct { + Gets, CacheHits, Loads, LocalLoads, PeerLoads, LocalLoadErrs int64 + MainBytes, MainItems, MainEvictions int64 + HotBytes, HotItems int64 +} + +// RegisterListingCache installs a Prometheus collector that reports the +// directory-listing cache's groupcache counters under objgit_s3_listing_cache_*. +// provider is polled at scrape time. Call once at startup when the cache is +// enabled. +func RegisterListingCache(provider func() ListingCacheStats) { + prometheus.MustRegister(&listingCacheCollector{provider: provider}) +} + +type listingCacheCollector struct { + provider func() ListingCacheStats +} + +var ( + lcGets = prometheus.NewDesc("objgit_s3_listing_cache_gets_total", "Listing-cache Get requests (incl. from peers).", nil, nil) + lcHits = prometheus.NewDesc("objgit_s3_listing_cache_hits_total", "Listing-cache requests served from cache.", nil, nil) + lcLoads = prometheus.NewDesc("objgit_s3_listing_cache_loads_total", "Listing-cache loads (gets minus hits).", nil, nil) + lcLocal = prometheus.NewDesc("objgit_s3_listing_cache_local_loads_total", "Listing-cache loads served by listing S3 locally.", nil, nil) + lcPeer = prometheus.NewDesc("objgit_s3_listing_cache_peer_loads_total", "Listing-cache loads served by a peer.", nil, nil) + lcLoadErrs = prometheus.NewDesc("objgit_s3_listing_cache_local_load_errors_total", "Listing-cache local load errors.", nil, nil) + lcBytes = prometheus.NewDesc("objgit_s3_listing_cache_bytes", "Listing-cache resident bytes by cache.", []string{"cache"}, nil) + lcItems = prometheus.NewDesc("objgit_s3_listing_cache_items", "Listing-cache resident items by cache.", []string{"cache"}, nil) + lcEvictions = prometheus.NewDesc("objgit_s3_listing_cache_evictions_total", "Listing-cache main-cache evictions.", nil, nil) +) + +func (c *listingCacheCollector) Describe(ch chan<- *prometheus.Desc) { + ch <- lcGets + ch <- lcHits + ch <- lcLoads + ch <- lcLocal + ch <- lcPeer + ch <- lcLoadErrs + ch <- lcBytes + ch <- lcItems + ch <- lcEvictions +} + +func (c *listingCacheCollector) Collect(ch chan<- prometheus.Metric) { + s := c.provider() + ch <- prometheus.MustNewConstMetric(lcGets, prometheus.CounterValue, float64(s.Gets)) + ch <- prometheus.MustNewConstMetric(lcHits, prometheus.CounterValue, float64(s.CacheHits)) + ch <- prometheus.MustNewConstMetric(lcLoads, prometheus.CounterValue, float64(s.Loads)) + ch <- prometheus.MustNewConstMetric(lcLocal, prometheus.CounterValue, float64(s.LocalLoads)) + ch <- prometheus.MustNewConstMetric(lcPeer, prometheus.CounterValue, float64(s.PeerLoads)) + ch <- prometheus.MustNewConstMetric(lcLoadErrs, prometheus.CounterValue, float64(s.LocalLoadErrs)) + ch <- prometheus.MustNewConstMetric(lcEvictions, prometheus.CounterValue, float64(s.MainEvictions)) + ch <- prometheus.MustNewConstMetric(lcBytes, prometheus.GaugeValue, float64(s.MainBytes), "main") + ch <- prometheus.MustNewConstMetric(lcBytes, prometheus.GaugeValue, float64(s.HotBytes), "hot") + ch <- prometheus.MustNewConstMetric(lcItems, prometheus.GaugeValue, float64(s.MainItems), "main") + ch <- prometheus.MustNewConstMetric(lcItems, prometheus.GaugeValue, float64(s.HotItems), "hot") +} diff --git a/internal/s3fs/basic.go b/internal/s3fs/basic.go index 53060f5..38241be 100644 --- a/internal/s3fs/basic.go +++ b/internal/s3fs/basic.go @@ -74,7 +74,27 @@ func (fs3 *S3FS) OpenFile(filename string, flag int, perm os.FileMode) (billy.Fi return &tempReadFile{buf: buf, name: filename}, nil } - f, err := newS3ReadFile(fs3.client, fs3.bucket, key, filename) + // If the parent folder's listing is cached, resolve the open without a + // negotiation round-trip: absent → not-exist, a sub-prefix → directory. + // For a present file the head cache (prewarmed in the background) lets + // newS3ReadFile skip its HeadObject; only GetObject fetches the body. + if info, found, known := fs3.resolve(context.TODO(), key); known { + switch { + case !found: + return nil, &os.PathError{Op: "open", Path: filename, Err: fs.ErrNotExist} + case info.Kind == kindDir: + return newS3DirFile(key, fs3.bucket, fs3.client), nil + default: + if ho, err := fs3.cache.headInfo(context.TODO(), key); err == nil { + return newS3ReadFile(fs3.client, fs3.bucket, key, filename, ho) + } else if isNotFound(err) { + return nil, &os.PathError{Op: "open", Path: filename, Err: fs.ErrNotExist} + } + // transient head-cache error: fall through to a live read. + } + } + + f, err := newS3ReadFile(fs3.client, fs3.bucket, key, filename, nil) if err == nil { return f, nil } @@ -111,16 +131,38 @@ func (fs3 *S3FS) OpenFile(filename string, flag int, perm os.FileMode) (billy.Fi return nil, &os.PathError{Op: "open", Path: filename, Err: fs.ErrNotExist} case O_WRONLY: - return newS3WriteFile(fs3.client, fs3.bucket, key, filename, fs3.unixMeta) + return newS3WriteFile(fs3.client, fs3.bucket, key, filename, fs3.unixMeta, fs3.cache) case O_WRMULTIPART: - return newS3MultipartUploadFile(fs3.client, fs3.bucket, key, filename, fs3.unixMeta) + return newS3MultipartUploadFile(fs3.client, fs3.bucket, key, filename, fs3.unixMeta, fs3.cache) default: return nil, errors.New("unsupported open flag") } } +// resolve consults the listing cache for key's parent folder. known is false +// when the cache is disabled or the lookup failed — the caller then falls back +// to a live S3 round-trip, so cache problems never fail an operation. When +// known, found reports whether base exists under the parent and info describes +// it (kind/size/mtime). +func (fs3 *S3FS) resolve(ctx context.Context, key string) (info childEntry, found, known bool) { + if fs3.cache == nil { + return childEntry{}, false, false + } + prefix, base := splitKey(key) + entries, err := fs3.cache.list(ctx, prefix) + if err != nil { + return childEntry{}, false, false + } + for _, e := range entries { + if e.Name == base { + return e, true, true + } + } + return childEntry{}, false, true +} + // Stat returns a FileInfo describing the named file. func (fs3 *S3FS) Stat(filename string) (os.FileInfo, error) { key := strings.TrimPrefix(fs3.cleanPath(filename), "/") @@ -136,6 +178,24 @@ func (fs3 *S3FS) Stat(filename string) (os.FileInfo, error) { ctx := context.TODO() + // If the parent folder's listing is cached, answer without a foreground S3 + // round-trip: absent → not-exist, a sub-prefix → directory, a present file + // → the head cache (prewarmed in the background from the listing). + if info, found, known := fs3.resolve(ctx, key); known { + if !found { + return nil, &os.PathError{Op: "stat", Path: filename, Err: fs.ErrNotExist} + } + if info.Kind == kindDir { + return newDirInfo(path.Base(key)), nil + } + if ho, err := fs3.cache.headInfo(ctx, key); err == nil { + return newFileInfoFromHead(path.Base(key), ho, fs3.unixMeta), nil + } else if isNotFound(err) { + return nil, &os.PathError{Op: "stat", Path: filename, Err: fs.ErrNotExist} + } + // transient head-cache error: fall through to a direct HeadObject. + } + start := time.Now() head, err := fs3.client.HeadObject(ctx, &s3.HeadObjectInput{ Bucket: &fs3.bucket, @@ -199,6 +259,10 @@ func (fs3 *S3FS) Rename(oldpath, newpath string) error { if err != nil { return fmt.Errorf("failed to upload temp %q to %q: %w", oldpath, newpath, err) } + if fs3.cache != nil { + prefix, _ := splitKey(dst) + fs3.cache.invalidate(prefix) + } return nil } @@ -217,6 +281,14 @@ func (fs3 *S3FS) Rename(oldpath, newpath string) error { return fmt.Errorf("failed to rename %q to %q: %w", oldpath, newpath, err) } + // Both folders changed: the source lost a child, the destination gained one. + if fs3.cache != nil { + srcPrefix, _ := splitKey(src) + dstPrefix, _ := splitKey(dst) + fs3.cache.invalidate(srcPrefix) + fs3.cache.invalidate(dstPrefix) + } + return nil } @@ -242,6 +314,10 @@ func (fs3 *S3FS) Remove(filename string) error { if err != nil { return fmt.Errorf("failed to remove file: %w", err) } + if fs3.cache != nil { + prefix, _ := splitKey(key) + fs3.cache.invalidate(prefix) + } return nil } diff --git a/internal/s3fs/chroot.go b/internal/s3fs/chroot.go index 5c7a601..cac6eeb 100644 --- a/internal/s3fs/chroot.go +++ b/internal/s3fs/chroot.go @@ -24,6 +24,7 @@ func (fs3 *S3FS) Chroot(path string) (billy.Filesystem, error) { root: p, separator: fs3.separator, unixMeta: fs3.unixMeta, + cache: fs3.cache, temps: make(map[string]*tempBuffer), } return nfs, nil diff --git a/internal/s3fs/dir.go b/internal/s3fs/dir.go index 4d7bce9..2acbcd0 100644 --- a/internal/s3fs/dir.go +++ b/internal/s3fs/dir.go @@ -15,27 +15,22 @@ import ( "github.com/aws/aws-sdk-go-v2/service/s3" ) -// ReadDir reads the directory named by dirname and returns a list of -// directory entries sorted by filename. -func (fs3 *S3FS) ReadDir(dir string) ([]fs.DirEntry, error) { - key := strings.TrimPrefix(fs3.cleanPath(dir), "/") - var prefix string - if key != "" && key != "." { - prefix = key + "/" - } - - ctx := context.TODO() - +// listChildren lists the immediate children of prefix (a full-canonical key +// prefix: "" for the bucket root, otherwise ending in separator), paginating to +// completion. Sub-prefixes come back as kindDir, objects as kindFile carrying +// size/mtime — dirs first then files, preserving S3's lexicographic order. It +// is a free function so the listing cache's getter, which holds only the raw +// client, can reuse it. +func listChildren(ctx context.Context, client s3Client, bucket, separator, prefix string) ([]childEntry, error) { var ct *string - var dirs []fs.DirEntry - var files []fs.DirEntry + var dirs, files []childEntry for { start := time.Now() - res, err := fs3.client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{ - Bucket: &fs3.bucket, + res, err := client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{ + Bucket: &bucket, Prefix: &prefix, ContinuationToken: ct, - Delimiter: &fs3.separator, + Delimiter: &separator, }) observeS3("ListObjectsV2", start, err) if err != nil { @@ -47,7 +42,7 @@ func (fs3 *S3FS) ReadDir(dir string) ([]fs.DirEntry, error) { if name == "" { continue } - dirs = append(dirs, fs.FileInfoToDirEntry(newDirInfo(name))) + dirs = append(dirs, childEntry{Name: name, Kind: kindDir}) } for _, f := range res.Contents { @@ -60,13 +55,12 @@ func (fs3 *S3FS) ReadDir(dir string) ([]fs.DirEntry, error) { if name == "" { continue } - files = append(files, - fs.FileInfoToDirEntry(newFileInfo( - pathpkg.Base(name), - aws.ToInt64(f.Size), - aws.ToTime(f.LastModified), - )), - ) + files = append(files, childEntry{ + Name: pathpkg.Base(name), + Kind: kindFile, + Size: aws.ToInt64(f.Size), + Mtime: aws.ToTime(f.LastModified).UnixNano(), + }) } if !aws.ToBool(res.IsTruncated) { @@ -78,6 +72,41 @@ func (fs3 *S3FS) ReadDir(dir string) ([]fs.DirEntry, error) { return append(dirs, files...), nil } +// ReadDir reads the directory named by dirname and returns a list of +// directory entries. When a listing cache is attached the listing is served +// through it (and reused by later Stat/Open of siblings). +func (fs3 *S3FS) ReadDir(dir string) ([]fs.DirEntry, error) { + key := strings.TrimPrefix(fs3.cleanPath(dir), "/") + var prefix string + if key != "" && key != "." { + prefix = key + "/" + } + + ctx := context.TODO() + + var entries []childEntry + var err error + if fs3.cache != nil { + entries, err = fs3.cache.list(ctx, prefix) + } else { + entries, err = listChildren(ctx, fs3.client, fs3.bucket, fs3.separator, prefix) + } + if err != nil { + return nil, err + } + + out := make([]fs.DirEntry, 0, len(entries)) + for _, e := range entries { + if e.Kind == kindDir { + out = append(out, fs.FileInfoToDirEntry(newDirInfo(e.Name))) + continue + } + out = append(out, fs.FileInfoToDirEntry(newFileInfo(e.Name, e.Size, time.Unix(0, e.Mtime)))) + } + + return out, nil +} + // MkdirAll creates a directory named path, along with any necessary // parents, and returns nil, or else returns an error. The permission bits // perm are used for all directories that MkdirAll creates. If path is/ @@ -90,6 +119,10 @@ func (fs3 *S3FS) MkdirAll(filename string, perm os.FileMode) error { Body: bytes.NewBuffer(nil), }) observeS3("PutObject", start, err) + if err == nil && fs3.cache != nil { + prefix, _ := splitKey(fs3.key(filename)) + fs3.cache.invalidate(prefix) + } return err } diff --git a/internal/s3fs/file.go b/internal/s3fs/file.go index 3de8453..11c1aaa 100644 --- a/internal/s3fs/file.go +++ b/internal/s3fs/file.go @@ -13,7 +13,6 @@ import ( "time" "github.com/aws/aws-sdk-go-v2/service/s3" - "github.com/tigrisdata/storage-go" "go.uber.org/atomic" "tangled.org/xeiaso.net/objgit/internal/s3fs/unixmeta" ) @@ -50,7 +49,7 @@ var ( // // Upon creation, the file is loaded from S3. type s3ReadFile struct { - client *storage.Client // S3 SDK client + client s3Client // S3 SDK client bucket string // S3 bucket name key string // File object's key in S3 name string // Root-relative path as presented to Open @@ -60,23 +59,29 @@ type s3ReadFile struct { } // newS3ReadFile creates a new s3ReadFile. key is the full S3 object key; name -// is the root-relative path the caller passed to Open (returned by Name). -func newS3ReadFile(client *storage.Client, bucket, key, name string) (*s3ReadFile, error) { +// is the root-relative path the caller passed to Open (returned by Name). When +// head is non-nil it is used as-is — the caller already has the object's +// metadata (e.g. from the listing-cache head precache) so the redundant +// HeadObject round-trip is skipped; pass nil to fetch it here. +func newS3ReadFile(client s3Client, bucket, key, name string, head *s3.HeadObjectOutput) (*s3ReadFile, error) { // Create the context ctx := context.TODO() // TODO: How can user-supplied contexts be supported? - start := time.Now() - ho, err := client.HeadObject(ctx, &s3.HeadObjectInput{ - Bucket: new(bucket), - Key: new(key), - }) - observeS3("HeadObject", start, err) - if err != nil { - return nil, &os.PathError{Op: "read", Path: key, Err: err} + if head == nil { + start := time.Now() + ho, err := client.HeadObject(ctx, &s3.HeadObjectInput{ + Bucket: new(bucket), + Key: new(key), + }) + observeS3("HeadObject", start, err) + if err != nil { + return nil, &os.PathError{Op: "read", Path: key, Err: err} + } + head = ho } // Run the GetObject operation - start = time.Now() + start := time.Now() res, err := client.GetObject(ctx, &s3.GetObjectInput{ Bucket: &bucket, Key: &key, @@ -100,7 +105,7 @@ func newS3ReadFile(client *storage.Client, bucket, key, name string) (*s3ReadFil key: key, name: name, reader: reader, - head: ho, + head: head, }, nil } @@ -179,18 +184,19 @@ func (f *s3ReadFile) Stat() (fs.FileInfo, error) { // Upon creation, a buffer is created to store the file contents. Upon close, // the file is uploaded to S3. type s3WriteFile struct { - client *storage.Client // s3 skd client + client s3Client // s3 skd client bucket string // S3 bucket name key string // File object's key in S3 name string // Root-relative path as presented to Open closed bool // Is the file closed? buf *bytes.Buffer // Buffer for storing the file before it's uploaded unixMeta *unixMetaConfig // optional POSIX attribute defaults (nil = disabled) + cache *ListingCache // optional listing cache to invalidate on upload } // newS3WriteFile creates a new s3WriteFile. key is the full S3 object key; name // is the root-relative path the caller passed to Open (returned by Name). -func newS3WriteFile(client *storage.Client, bucket, key, name string, cfg *unixMetaConfig) (*s3WriteFile, error) { +func newS3WriteFile(client s3Client, bucket, key, name string, cfg *unixMetaConfig, cache *ListingCache) (*s3WriteFile, error) { return &s3WriteFile{ client: client, bucket: bucket, @@ -198,6 +204,7 @@ func newS3WriteFile(client *storage.Client, bucket, key, name string, cfg *unixM name: name, buf: bytes.NewBuffer(nil), unixMeta: cfg, + cache: cache, }, nil } @@ -262,6 +269,13 @@ func (f *s3WriteFile) Close() error { return fmt.Errorf("unable to perform GetObject operation: %w", err) } + // The new object changes its parent folder's listing; drop the cached + // listing so the next read re-lists and sees it. + if f.cache != nil { + prefix, _ := splitKey(f.key) + f.cache.invalidate(prefix) + } + return nil } @@ -287,18 +301,19 @@ func (f *s3WriteFile) Truncate(size int64) error { // s3MultipartUploadFile implements billy.File type s3MultipartUploadFile struct { - client *storage.Client // s3 skd client - bucket string // S3 bucket name - key string // File object's key in S3 - name string // Root-relative path as presented to Open - closed bool // Is the file closed? - uploadID string // S3 multipart upload ID - uploadN *atomic.Int32 // Counter tracking the number of uploads + client s3Client // s3 skd client + bucket string // S3 bucket name + key string // File object's key in S3 + name string // Root-relative path as presented to Open + closed bool // Is the file closed? + uploadID string // S3 multipart upload ID + uploadN *atomic.Int32 // Counter tracking the number of uploads + cache *ListingCache // optional listing cache to invalidate on upload } // newS3MultipartUploadFile creates a new s3MultipartUploadFile. key is the full // S3 object key; name is the root-relative path passed to Open. -func newS3MultipartUploadFile(client *storage.Client, bucket, key, name string, cfg *unixMetaConfig) (*s3MultipartUploadFile, error) { +func newS3MultipartUploadFile(client s3Client, bucket, key, name string, cfg *unixMetaConfig, cache *ListingCache) (*s3MultipartUploadFile, error) { // TODO: Check if the file exists // ... @@ -326,6 +341,7 @@ func newS3MultipartUploadFile(client *storage.Client, bucket, key, name string, name: name, uploadID: *res.UploadId, uploadN: atomic.NewInt32(1), + cache: cache, }, nil } @@ -412,6 +428,11 @@ func (f *s3MultipartUploadFile) Close() error { return fmt.Errorf("unable to complete multipart upload: %w", err) } + if f.cache != nil { + prefix, _ := splitKey(f.key) + f.cache.invalidate(prefix) + } + return nil } @@ -430,10 +451,10 @@ func (f *s3MultipartUploadFile) Stat() (fs.FileInfo, error) { type s3DirFile struct { name, bucket string closed bool - cli *storage.Client + cli s3Client } -func newS3DirFile(name, bucket string, cli *storage.Client) *s3DirFile { +func newS3DirFile(name, bucket string, cli s3Client) *s3DirFile { return &s3DirFile{ name: name, bucket: bucket, diff --git a/internal/s3fs/filesystem.go b/internal/s3fs/filesystem.go index 74d323c..1877a78 100644 --- a/internal/s3fs/filesystem.go +++ b/internal/s3fs/filesystem.go @@ -1,20 +1,36 @@ package s3fs import ( + "context" "fmt" "os" "path" "strings" "sync" + "github.com/aws/aws-sdk-go-v2/service/s3" "github.com/go-git/go-billy/v6" - "github.com/tigrisdata/storage-go" ) const ( DefaultSeparator = "/" ) +// s3Client is the subset of *storage.Client (which embeds *s3.Client) that the +// filesystem uses. Naming it as an interface lets tests substitute a counting +// stub; the concrete Tigris client satisfies it unchanged. +type s3Client interface { + HeadObject(context.Context, *s3.HeadObjectInput, ...func(*s3.Options)) (*s3.HeadObjectOutput, error) + GetObject(context.Context, *s3.GetObjectInput, ...func(*s3.Options)) (*s3.GetObjectOutput, error) + PutObject(context.Context, *s3.PutObjectInput, ...func(*s3.Options)) (*s3.PutObjectOutput, error) + ListObjectsV2(context.Context, *s3.ListObjectsV2Input, ...func(*s3.Options)) (*s3.ListObjectsV2Output, error) + DeleteObject(context.Context, *s3.DeleteObjectInput, ...func(*s3.Options)) (*s3.DeleteObjectOutput, error) + RenameObject(context.Context, *s3.CopyObjectInput, ...func(*s3.Options)) (*s3.CopyObjectOutput, error) + CreateMultipartUpload(context.Context, *s3.CreateMultipartUploadInput, ...func(*s3.Options)) (*s3.CreateMultipartUploadOutput, error) + UploadPart(context.Context, *s3.UploadPartInput, ...func(*s3.Options)) (*s3.UploadPartOutput, error) + CompleteMultipartUpload(context.Context, *s3.CompleteMultipartUploadInput, ...func(*s3.Options)) (*s3.CompleteMultipartUploadOutput, error) +} + // unixMetaConfig holds the session defaults used when the optional Unix-metadata // feature is enabled. A nil *unixMetaConfig means the feature is off and the // filesystem behaves as if no POSIX attributes exist. @@ -24,12 +40,17 @@ type unixMetaConfig struct { } type S3FS struct { - client *storage.Client + client s3Client bucket string root string separator string unixMeta *unixMetaConfig + // cache, when non-nil, memoises directory listings so Stat/Open of a path + // whose parent folder has been listed can skip the S3 round-trip. It is + // shared by pointer across this filesystem and all of its Chroot children. + cache *ListingCache + // temps holds TempFile-backed buffers keyed by canonical S3 key, so a // subsequent Open of the same path returns a reader over the same bytes // the writer is still appending to. See tempfs.go. @@ -51,8 +72,18 @@ func WithUnixMetadata(uid, gid uint32, umask os.FileMode) Option { } } -// NewS3FS creates a new S3FS Filesystem. -func NewS3FS(client *storage.Client, bucket string, opts ...Option) (billy.Filesystem, error) { +// WithListingCache attaches a directory-listing cache. The same *ListingCache is +// carried into every Chroot child so the whole tree shares one cache keyed by +// full-canonical prefix. Construct the cache with NewListingCache. +func WithListingCache(c *ListingCache) Option { + return func(fs3 *S3FS) { + fs3.cache = c + } +} + +// NewS3FS creates a new S3FS Filesystem. client is typically a *storage.Client; +// it is accepted as the s3Client interface so tests can substitute a stub. +func NewS3FS(client s3Client, bucket string, opts ...Option) (billy.Filesystem, error) { // Check for a non-nil client if client == nil { return nil, fmt.Errorf("s3 client cannot be nil") diff --git a/internal/s3fs/listingcache.go b/internal/s3fs/listingcache.go new file mode 100644 index 0000000..ccfa69e --- /dev/null +++ b/internal/s3fs/listingcache.go @@ -0,0 +1,392 @@ +// listingcache.go caches directory listings so that Stat/Open of a path whose +// parent folder has been listed can be answered without an S3 round-trip: an +// absent child returns "not found" for free, a sub-directory is recognised for +// free, and only a present file still issues a HeadObject/GetObject for its +// authoritative content/metadata. +// +// The store is github.com/golang/groupcache so the listing can be shared across +// a fleet via its consistent-hash peer pool. groupcache is fill-once and +// immutable (no Set, Delete, or TTL), so two facts are encoded into the cache +// key instead: +// +// - a time window floor(now/TTL): when it advances the key changes, forcing a +// re-list. This bounds staleness from writers this process cannot see. +// - a per-prefix local generation, bumped on every local write under the +// prefix: this moves the key so the next local read re-lists and sees the +// write immediately (read-after-write correctness groupcache alone cannot +// give). The stale entry under the old key is simply never queried again. +package s3fs + +import ( + "context" + "encoding/json" + "errors" + "log/slog" + "net/http" + "strconv" + "strings" + "sync" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/aws/smithy-go" + "github.com/golang/groupcache" +) + +// maxPrefetchInFlight bounds the background HeadObject precaches in flight at +// once, so listing a large folder can't spawn an unbounded goroutine/S3 storm. +// Overflow files are simply fetched on demand instead. +const maxPrefetchInFlight = 64 + +// childKind distinguishes the two things a listing can report under a prefix. +type childKind uint8 + +const ( + kindFile childKind = iota + kindDir +) + +// childEntry is one immediate child of a listed prefix. name is the child's +// base name (no separator). size/mtime are populated for files only; dirs +// (S3 common prefixes) carry zero values. +type childEntry struct { + Name string `json:"n"` + Kind childKind `json:"k"` + Size int64 `json:"s,omitempty"` + Mtime int64 `json:"m,omitempty"` // unix nanoseconds +} + +// headData is the cached result of a HeadObject: everything the file metadata +// helpers need. Meta carries the x-amz-meta-* user metadata for the +// Unix-metadata feature (listings can't, which is why heads are cached too). +type headData struct { + Size int64 `json:"s"` + Mtime int64 `json:"m"` // unix nanoseconds + Meta map[string]string `json:"u,omitempty"` +} + +// DefaultGroupName is the groupcache group every objgitd uses; it must match +// across peers for cross-process sharing to route correctly. +const DefaultGroupName = "objgit-listings" + +// CacheConfig configures a ListingCache. TTL must be > 0 (callers gate the whole +// feature on it). The remaining fields have sane zero-value behaviour. +type CacheConfig struct { + TTL time.Duration // key window; bounds cross-process staleness + RefreshInterval time.Duration // warmer tick; <=0 disables the warmer + IdleTTL time.Duration // drop un-accessed prefixes from the warmer + SizeBytes int64 // groupcache LRU budget (<=0 → 64 MiB) + Name string // groupcache group name (default DefaultGroupName) + + // DisableHeadPrefetch turns off the background HeadObject precache that + // warms the head cache for every file in a listing. The zero value keeps + // prefetching enabled. + DisableHeadPrefetch bool + + Self string // this node's groupcache base URL ("" → single-process) + Peers []string // peer base URLs, including Self, when sharing +} + +// ListingCache wraps a groupcache group with the window/generation key scheme +// and a background warmer. It is safe for concurrent use and is shared by +// pointer across an S3FS and all of its Chroot children. +type ListingCache struct { + group *groupcache.Group + headGroup *groupcache.Group // per-object HeadObject metadata + pool *groupcache.HTTPPool // nil in single-process mode + ttl time.Duration + cfg CacheConfig + + clock func() time.Time // overridable in tests + prefetchSem chan struct{} // bounds background head precaches + + mu sync.Mutex + gens map[string]uint64 // per-prefix local generation + seen map[string]time.Time // prefixes accessed → driven by the warmer +} + +// NewListingCache builds a cache backed by a groupcache group whose getter lists +// the requested prefix from S3. client/bucket/separator are captured for the +// getter; they are the root filesystem's, so cached prefixes are full-canonical +// keys that every Chroot view agrees on. +func NewListingCache(cfg CacheConfig, client s3Client, bucket, separator string) *ListingCache { + if cfg.Name == "" { + cfg.Name = DefaultGroupName + } + if cfg.SizeBytes <= 0 { + cfg.SizeBytes = 64 << 20 + } + c := &ListingCache{ + ttl: cfg.TTL, + cfg: cfg, + clock: time.Now, + prefetchSem: make(chan struct{}, maxPrefetchInFlight), + gens: map[string]uint64{}, + seen: map[string]time.Time{}, + } + + listGetter := groupcache.GetterFunc(func(ctx context.Context, key string, dest groupcache.Sink) error { + prefix := keyPrefix(key) + entries, err := listChildren(ctx, client, bucket, separator, prefix) + if err != nil { + return err + } + data, err := json.Marshal(entries) + if err != nil { + return err + } + // Precache each file's HeadObject in the background, off this request's + // critical path; groupcache singleflight coalesces a background precache + // with any foreground head lookup for the same object. + c.prefetchHeads(prefix, entries) + return dest.SetBytes(data) + }) + c.group = groupcache.NewGroup(cfg.Name, cfg.SizeBytes, listGetter) + + headGetter := groupcache.GetterFunc(func(ctx context.Context, key string, dest groupcache.Sink) error { + objKey := keyPrefix(key) + start := time.Now() + ho, err := client.HeadObject(ctx, &s3.HeadObjectInput{Bucket: &bucket, Key: &objKey}) + observeS3("HeadObject", start, err) + if err != nil { + return err + } + data, err := json.Marshal(headData{ + Size: aws.ToInt64(ho.ContentLength), + Mtime: aws.ToTime(ho.LastModified).UnixNano(), + Meta: ho.Metadata, + }) + if err != nil { + return err + } + return dest.SetBytes(data) + }) + c.headGroup = groupcache.NewGroup(cfg.Name+"-heads", cfg.SizeBytes, headGetter) + + if cfg.Self != "" { + c.pool = groupcache.NewHTTPPoolOpts(cfg.Self, &groupcache.HTTPPoolOptions{}) + peers := cfg.Peers + if len(peers) == 0 { + peers = []string{cfg.Self} + } + c.pool.Set(peers...) + } + + return c +} + +// list returns the immediate children of prefix, served from groupcache (which +// fills via S3 on a miss). prefix is a full-canonical key prefix: "" for the +// bucket root, otherwise ending in the separator. +func (c *ListingCache) list(ctx context.Context, prefix string) ([]childEntry, error) { + c.touch(prefix) + + var data []byte + if err := c.group.Get(ctx, c.groupKey(prefix), groupcache.AllocatingByteSliceSink(&data)); err != nil { + return nil, err + } + var entries []childEntry + if err := json.Unmarshal(data, &entries); err != nil { + return nil, err + } + return entries, nil +} + +// groupKey encodes the prefix, the current time window, and the prefix's local +// generation. The prefix is first so the getter can recover it. +func (c *ListingCache) groupKey(prefix string) string { + window := c.clock().UnixNano() / int64(c.ttl) + return prefix + "\x00" + strconv.FormatInt(window, 10) + "\x00" + strconv.FormatUint(c.gen(prefix), 10) +} + +// headKey is groupKey's analogue for the per-object head cache. It carries the +// object key plus the same window and its parent prefix's generation, so a +// write under that prefix (which bumps the generation) re-heads the object. +func (c *ListingCache) headKey(objKey string) string { + prefix, _ := splitKey(objKey) + window := c.clock().UnixNano() / int64(c.ttl) + return objKey + "\x00" + strconv.FormatInt(window, 10) + "\x00" + strconv.FormatUint(c.gen(prefix), 10) +} + +// keyPrefix recovers the object key / prefix a getter was asked for, stripping +// the "\x00window\x00gen" suffix groupKey/headKey append. +func keyPrefix(key string) string { + if i := strings.IndexByte(key, 0); i >= 0 { + return key[:i] + } + return key +} + +// headInfo returns the object's HeadObject metadata, served from the head cache +// (prewarmed in the background from listings, or filled on demand here). +func (c *ListingCache) headInfo(ctx context.Context, objKey string) (*s3.HeadObjectOutput, error) { + var data []byte + if err := c.headGroup.Get(ctx, c.headKey(objKey), groupcache.AllocatingByteSliceSink(&data)); err != nil { + return nil, err + } + var d headData + if err := json.Unmarshal(data, &d); err != nil { + return nil, err + } + return &s3.HeadObjectOutput{ + ContentLength: aws.Int64(d.Size), + LastModified: aws.Time(time.Unix(0, d.Mtime)), + Metadata: d.Meta, + }, nil +} + +// prefetchHeads warms the head cache for every file in a freshly-listed folder, +// each via a detached background goroutine so the work never blocks the request +// that triggered the listing. Beyond maxPrefetchInFlight concurrent precaches it +// drops extras (logged), which are then fetched on demand. +func (c *ListingCache) prefetchHeads(prefix string, entries []childEntry) { + if c.cfg.DisableHeadPrefetch { + return + } + for _, e := range entries { + if e.Kind != kindFile { + continue + } + objKey := prefix + e.Name + select { + case c.prefetchSem <- struct{}{}: + default: + slog.Debug("head precache skipped: prefetch pipeline full", "key", objKey) + continue + } + go func(k string) { + defer func() { <-c.prefetchSem }() + var data []byte + if err := c.headGroup.Get(context.Background(), c.headKey(k), groupcache.AllocatingByteSliceSink(&data)); err != nil { + slog.Debug("head precache failed", "key", k, "err", err) + } + }(objKey) + } +} + +// isNotFound reports whether err is an S3 "object does not exist" error. +func isNotFound(err error) bool { + var apiErr smithy.APIError + if errors.As(err, &apiErr) { + switch apiErr.ErrorCode() { + case "NotFound", "NoSuchKey": + return true + } + } + return false +} + +func (c *ListingCache) gen(prefix string) uint64 { + c.mu.Lock() + defer c.mu.Unlock() + return c.gens[prefix] +} + +// invalidate bumps a prefix's local generation so the next read in this process +// re-lists. groupcache cannot delete; moving the key is how we invalidate. +func (c *ListingCache) invalidate(prefix string) { + if c == nil { + return + } + c.mu.Lock() + c.gens[prefix]++ + c.seen[prefix] = c.clock() + c.mu.Unlock() +} + +func (c *ListingCache) touch(prefix string) { + c.mu.Lock() + c.seen[prefix] = c.clock() + c.mu.Unlock() +} + +// RunWarmer pre-fills the current-window key for every recently-accessed prefix +// on each tick, smoothing window-rollover misses, and evicts prefixes unused for +// longer than IdleTTL. It returns when ctx is cancelled; it is a no-op when the +// refresh interval is non-positive. +func (c *ListingCache) RunWarmer(ctx context.Context) { + if c == nil || c.cfg.RefreshInterval <= 0 { + <-ctx.Done() + return + } + t := time.NewTicker(c.cfg.RefreshInterval) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + c.warmOnce(ctx) + } + } +} + +func (c *ListingCache) warmOnce(ctx context.Context) { + now := c.clock() + c.mu.Lock() + prefixes := make([]string, 0, len(c.seen)) + for p, last := range c.seen { + if c.cfg.IdleTTL > 0 && now.Sub(last) > c.cfg.IdleTTL { + delete(c.seen, p) + continue + } + prefixes = append(prefixes, p) + } + c.mu.Unlock() + + for _, p := range prefixes { + var data []byte + if err := c.group.Get(ctx, c.groupKey(p), groupcache.AllocatingByteSliceSink(&data)); err != nil { + slog.Debug("listing-cache warm failed", "prefix", p, "err", err) + } + } +} + +// PoolHandler returns the groupcache peer HTTP handler, or nil in single-process +// mode. Serve it on the address passed as -groupcache-bind. +func (c *ListingCache) PoolHandler() http.Handler { + if c == nil || c.pool == nil { + return nil + } + return c.pool +} + +// CacheStats is a flat snapshot of groupcache counters for export to metrics. +type CacheStats struct { + Gets, CacheHits, Loads, LocalLoads, PeerLoads, LocalLoadErrs int64 + MainBytes, MainItems, MainEvictions int64 + HotBytes, HotItems int64 +} + +// Stats snapshots the underlying groupcache counters. +func (c *ListingCache) Stats() CacheStats { + if c == nil || c.group == nil { + return CacheStats{} + } + main := c.group.CacheStats(groupcache.MainCache) + hot := c.group.CacheStats(groupcache.HotCache) + return CacheStats{ + Gets: c.group.Stats.Gets.Get(), + CacheHits: c.group.Stats.CacheHits.Get(), + Loads: c.group.Stats.Loads.Get(), + LocalLoads: c.group.Stats.LocalLoads.Get(), + PeerLoads: c.group.Stats.PeerLoads.Get(), + LocalLoadErrs: c.group.Stats.LocalLoadErrs.Get(), + MainBytes: main.Bytes, + MainItems: main.Items, + MainEvictions: main.Evictions, + HotBytes: hot.Bytes, + HotItems: hot.Items, + } +} + +// splitKey splits a canonical key into its parent prefix and base name. The +// prefix matches ReadDir's convention: "" for a root-level key, otherwise it +// ends in "/". +func splitKey(key string) (prefix, base string) { + if i := strings.LastIndex(key, "/"); i >= 0 { + return key[:i+1], key[i+1:] + } + return "", key +} diff --git a/internal/s3fs/listingcache_test.go b/internal/s3fs/listingcache_test.go new file mode 100644 index 0000000..bccbdf9 --- /dev/null +++ b/internal/s3fs/listingcache_test.go @@ -0,0 +1,383 @@ +package s3fs + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "io/fs" + "sort" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/aws/aws-sdk-go-v2/service/s3/types" + "github.com/aws/smithy-go" + "github.com/go-git/go-billy/v6" +) + +// stubClient is a counting, in-memory s3Client for cache tests. It honours +// Prefix+Delimiter listing so ReadDir/Stat resolution behaves like S3. +type stubClient struct { + mu sync.Mutex + keys map[string]int64 // object key -> size + + heads atomic.Int64 + lists atomic.Int64 + puts atomic.Int64 +} + +func newStub(keys ...string) *stubClient { + s := &stubClient{keys: map[string]int64{}} + for _, k := range keys { + s.keys[k] = 0 + } + return s +} + +func notFound() error { return &smithy.GenericAPIError{Code: "NotFound", Message: "not found"} } + +func (s *stubClient) HeadObject(_ context.Context, in *s3.HeadObjectInput, _ ...func(*s3.Options)) (*s3.HeadObjectOutput, error) { + s.heads.Add(1) + s.mu.Lock() + size, ok := s.keys[aws.ToString(in.Key)] + s.mu.Unlock() + if !ok { + return nil, notFound() + } + return &s3.HeadObjectOutput{ContentLength: aws.Int64(size), LastModified: aws.Time(time.Unix(0, 0))}, nil +} + +func (s *stubClient) ListObjectsV2(_ context.Context, in *s3.ListObjectsV2Input, _ ...func(*s3.Options)) (*s3.ListObjectsV2Output, error) { + s.lists.Add(1) + prefix := aws.ToString(in.Prefix) + delim := aws.ToString(in.Delimiter) + + s.mu.Lock() + ks := make([]string, 0, len(s.keys)) + for k := range s.keys { + ks = append(ks, k) + } + sizes := make(map[string]int64, len(s.keys)) + for k, v := range s.keys { + sizes[k] = v + } + s.mu.Unlock() + sort.Strings(ks) + + seenCP := map[string]bool{} + var cps []types.CommonPrefix + var contents []types.Object + for _, k := range ks { + if !strings.HasPrefix(k, prefix) { + continue + } + rest := k[len(prefix):] + if delim != "" { + if i := strings.Index(rest, delim); i >= 0 { + cp := prefix + rest[:i+1] + if !seenCP[cp] { + seenCP[cp] = true + cps = append(cps, types.CommonPrefix{Prefix: aws.String(cp)}) + } + continue + } + } + contents = append(contents, types.Object{ + Key: aws.String(k), + Size: aws.Int64(sizes[k]), + LastModified: aws.Time(time.Unix(0, 0)), + }) + } + return &s3.ListObjectsV2Output{Contents: contents, CommonPrefixes: cps, IsTruncated: aws.Bool(false)}, nil +} + +func (s *stubClient) PutObject(_ context.Context, in *s3.PutObjectInput, _ ...func(*s3.Options)) (*s3.PutObjectOutput, error) { + s.puts.Add(1) + s.mu.Lock() + s.keys[aws.ToString(in.Key)] = 0 + s.mu.Unlock() + return &s3.PutObjectOutput{}, nil +} + +func (s *stubClient) GetObject(_ context.Context, in *s3.GetObjectInput, _ ...func(*s3.Options)) (*s3.GetObjectOutput, error) { + return &s3.GetObjectOutput{Body: io.NopCloser(bytes.NewReader(nil))}, nil +} + +func (s *stubClient) DeleteObject(_ context.Context, in *s3.DeleteObjectInput, _ ...func(*s3.Options)) (*s3.DeleteObjectOutput, error) { + s.mu.Lock() + delete(s.keys, aws.ToString(in.Key)) + s.mu.Unlock() + return &s3.DeleteObjectOutput{}, nil +} + +func (s *stubClient) RenameObject(_ context.Context, in *s3.CopyObjectInput, _ ...func(*s3.Options)) (*s3.CopyObjectOutput, error) { + return &s3.CopyObjectOutput{}, nil +} + +func (s *stubClient) CreateMultipartUpload(_ context.Context, in *s3.CreateMultipartUploadInput, _ ...func(*s3.Options)) (*s3.CreateMultipartUploadOutput, error) { + return &s3.CreateMultipartUploadOutput{UploadId: aws.String("u")}, nil +} + +func (s *stubClient) UploadPart(_ context.Context, in *s3.UploadPartInput, _ ...func(*s3.Options)) (*s3.UploadPartOutput, error) { + return &s3.UploadPartOutput{}, nil +} + +func (s *stubClient) CompleteMultipartUpload(_ context.Context, in *s3.CompleteMultipartUploadInput, _ ...func(*s3.Options)) (*s3.CompleteMultipartUploadOutput, error) { + return &s3.CompleteMultipartUploadOutput{}, nil +} + +// gcSeq gives every test cache a unique groupcache group name; NewGroup panics +// on a duplicate name. +var gcSeq atomic.Int64 + +// newTestCache disables the background head precache so S3-call counts are +// deterministic; TestListingCacheHeadPrefetch exercises it explicitly. +func newTestCache(stub *stubClient, ttl time.Duration) *ListingCache { + n := gcSeq.Add(1) + return NewListingCache(CacheConfig{TTL: ttl, Name: fmt.Sprintf("test-%d", n), DisableHeadPrefetch: true}, stub, "bucket", "/") +} + +func newCachedFS(t *testing.T, stub *stubClient, cache *ListingCache) billy.Filesystem { + t.Helper() + fsys, err := NewS3FS(stub, "bucket", WithListingCache(cache)) + if err != nil { + t.Fatalf("NewS3FS: %v", err) + } + return fsys +} + +func TestListingCachePopulateOnMiss(t *testing.T) { + stub := newStub("objects/ab/file1") + cache := newTestCache(stub, time.Hour) + fsys := newCachedFS(t, stub, cache) + + // First Stat of an absent key lists the parent once, no HeadObject. + if _, err := fsys.Stat("objects/ab/nope"); !errors.Is(err, fs.ErrNotExist) { + t.Fatalf("Stat absent: want ErrNotExist, got %v", err) + } + if got := stub.lists.Load(); got != 1 { + t.Fatalf("first miss: lists = %d, want 1", got) + } + if got := stub.heads.Load(); got != 0 { + t.Fatalf("first miss: heads = %d, want 0", got) + } + + // A second absent sibling is a pure negative hit: no S3 at all. + l0, h0 := stub.lists.Load(), stub.heads.Load() + if _, err := fsys.Stat("objects/ab/nope2"); !errors.Is(err, fs.ErrNotExist) { + t.Fatalf("Stat absent2: want ErrNotExist, got %v", err) + } + if stub.lists.Load() != l0 || stub.heads.Load() != h0 { + t.Fatalf("negative hit did S3: lists %d->%d heads %d->%d", l0, stub.lists.Load(), h0, stub.heads.Load()) + } + + // A present file resolves from cache but still HeadObjects for metadata. + l0, h0 = stub.lists.Load(), stub.heads.Load() + fi, err := fsys.Stat("objects/ab/file1") + if err != nil { + t.Fatalf("Stat present: %v", err) + } + if fi.IsDir() { + t.Fatalf("Stat present: reported a directory") + } + if stub.lists.Load() != l0 { + t.Fatalf("present file re-listed: %d->%d", l0, stub.lists.Load()) + } + if stub.heads.Load() != h0+1 { + t.Fatalf("present file heads = %d, want %d", stub.heads.Load(), h0+1) + } +} + +func TestListingCacheReadDirThenStat(t *testing.T) { + stub := newStub("objects/ab/file1", "objects/ab/sub/deep") + cache := newTestCache(stub, time.Hour) + fsys := newCachedFS(t, stub, cache) + + entries, err := fsys.ReadDir("objects/ab") + if err != nil { + t.Fatalf("ReadDir: %v", err) + } + if len(entries) != 2 { + t.Fatalf("ReadDir entries = %d, want 2", len(entries)) + } + if stub.lists.Load() != 1 { + t.Fatalf("ReadDir lists = %d, want 1", stub.lists.Load()) + } + + // The sub-directory resolves from the cached listing with no S3 calls. + l0, h0 := stub.lists.Load(), stub.heads.Load() + fi, err := fsys.Stat("objects/ab/sub") + if err != nil { + t.Fatalf("Stat dir: %v", err) + } + if !fi.IsDir() { + t.Fatalf("Stat dir: not a directory") + } + if stub.lists.Load() != l0 || stub.heads.Load() != h0 { + t.Fatalf("dir resolve did S3: lists %d->%d heads %d->%d", l0, stub.lists.Load(), h0, stub.heads.Load()) + } + + // And an absent sibling is a negative hit. + if _, err := fsys.Stat("objects/ab/missing"); !errors.Is(err, fs.ErrNotExist) { + t.Fatalf("Stat missing: want ErrNotExist, got %v", err) + } + if stub.lists.Load() != l0 || stub.heads.Load() != h0 { + t.Fatalf("negative hit did S3 after ReadDir") + } +} + +func TestListingCacheInvalidatesOnWrite(t *testing.T) { + stub := newStub("refs/heads/main") + cache := newTestCache(stub, time.Hour) + fsys := newCachedFS(t, stub, cache) + + // Warm the cache and confirm the new ref reads as absent. + if _, err := fsys.Stat("refs/heads/new"); !errors.Is(err, fs.ErrNotExist) { + t.Fatalf("Stat before write: want ErrNotExist, got %v", err) + } + + // Write the ref; Close must invalidate the parent prefix. + f, err := fsys.Create("refs/heads/new") + if err != nil { + t.Fatalf("Create: %v", err) + } + if _, err := f.Write([]byte("deadbeef")); err != nil { + t.Fatalf("Write: %v", err) + } + if err := f.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + // The next Stat must re-list (generation bumped) and now see the ref. + fi, err := fsys.Stat("refs/heads/new") + if err != nil { + t.Fatalf("Stat after write: %v", err) + } + if fi.IsDir() { + t.Fatalf("Stat after write: reported a directory") + } +} + +func TestListingCacheWindowExpiry(t *testing.T) { + stub := newStub("objects/ab/file1") + cache := newTestCache(stub, time.Minute) + now := time.Unix(1_000_000, 0) + cache.clock = func() time.Time { return now } + fsys := newCachedFS(t, stub, cache) + + if _, err := fsys.Stat("objects/ab/nope"); !errors.Is(err, fs.ErrNotExist) { + t.Fatal("Stat 1") + } + if stub.lists.Load() != 1 { + t.Fatalf("lists = %d, want 1", stub.lists.Load()) + } + // Same window: served from cache. + if _, err := fsys.Stat("objects/ab/nope"); !errors.Is(err, fs.ErrNotExist) { + t.Fatal("Stat 2") + } + if stub.lists.Load() != 1 { + t.Fatalf("same window re-listed: lists = %d, want 1", stub.lists.Load()) + } + // Advance past the TTL window: the key changes and we re-list. + now = now.Add(2 * time.Minute) + if _, err := fsys.Stat("objects/ab/nope"); !errors.Is(err, fs.ErrNotExist) { + t.Fatal("Stat 3") + } + if stub.lists.Load() != 2 { + t.Fatalf("new window lists = %d, want 2", stub.lists.Load()) + } +} + +func TestListingCacheWarmer(t *testing.T) { + stub := newStub("objects/ab/file1") + cache := newTestCache(stub, time.Hour) + cache.cfg.RefreshInterval = time.Millisecond // enable warming + cache.cfg.IdleTTL = time.Minute + now := time.Unix(1_000_000, 0) + cache.clock = func() time.Time { return now } + + // Touch a prefix, then warm: the warmer fills it (one list). + cache.touch("objects/ab/") + cache.warmOnce(context.Background()) + if stub.lists.Load() != 1 { + t.Fatalf("warmer lists = %d, want 1", stub.lists.Load()) + } + + // After the idle window the warmer evicts the prefix from its working set. + now = now.Add(2 * time.Minute) + cache.warmOnce(context.Background()) + cache.mu.Lock() + n := len(cache.seen) + cache.mu.Unlock() + if n != 0 { + t.Fatalf("idle prefix not evicted: seen = %d", n) + } +} + +func TestListingCacheHeadPrefetch(t *testing.T) { + stub := newStub("objects/ab/f1", "objects/ab/f2") + n := gcSeq.Add(1) + cache := NewListingCache(CacheConfig{TTL: time.Hour, Name: fmt.Sprintf("test-%d", n)}, stub, "bucket", "/") + fsys := newCachedFS(t, stub, cache) + + // A single listing fill (here triggered by an absent lookup) prefetches the + // HeadObject for every file in the folder, in the background. + if _, err := fsys.Stat("objects/ab/nope"); !errors.Is(err, fs.ErrNotExist) { + t.Fatalf("Stat: %v", err) + } + + // Wait for the two background HeadObjects to land. + deadline := time.Now().Add(2 * time.Second) + for stub.heads.Load() < 2 { + if time.Now().After(deadline) { + t.Fatalf("prefetch did not warm both heads: got %d", stub.heads.Load()) + } + time.Sleep(time.Millisecond) + } + if got := stub.heads.Load(); got != 2 { + t.Fatalf("prefetch heads = %d, want 2", got) + } + + // Stat of a prewarmed file is served from the head cache: no new HeadObject. + h0 := stub.heads.Load() + if _, err := fsys.Stat("objects/ab/f1"); err != nil { + t.Fatalf("Stat f1: %v", err) + } + if stub.heads.Load() != h0 { + t.Fatalf("prewarmed Stat did a HeadObject: %d->%d", h0, stub.heads.Load()) + } +} + +func TestListingCacheChrootShares(t *testing.T) { + stub := newStub("repo/objects/ab/file1") + cache := newTestCache(stub, time.Hour) + rootfs := newCachedFS(t, stub, cache) + + // Populate the listing for repo/objects/ab/ via the root view. + if _, err := rootfs.Stat("repo/objects/ab/missing1"); !errors.Is(err, fs.ErrNotExist) { + t.Fatalf("root Stat: want ErrNotExist, got %v", err) + } + if stub.lists.Load() != 1 { + t.Fatalf("root Stat lists = %d, want 1", stub.lists.Load()) + } + + // A chroot view shares the same cache keyed by canonical prefix, so a Stat + // under it hits the cached listing with no further S3. + sub, err := rootfs.Chroot("repo") + if err != nil { + t.Fatalf("Chroot: %v", err) + } + l0 := stub.lists.Load() + if _, err := sub.Stat("objects/ab/missing2"); !errors.Is(err, fs.ErrNotExist) { + t.Fatalf("chroot Stat: want ErrNotExist, got %v", err) + } + if stub.lists.Load() != l0 { + t.Fatalf("chroot Stat re-listed: %d->%d", l0, stub.lists.Load()) + } +}