feat: keep pushed packs whole · xeiaso.net/objgit@ab60b2d

+4 -2

CLAUDE.md

··· 48 48 49 49 ### Two subtle protocol points 50 50 51 - 1. **`streamingStorer` in `git_protocol.go`** wraps the storer for git:// receive-pack to **hide its `PackfileWriter` capability**. `UpdateObjectStorage`'s `PackfileWriter` path uses `io.CopyBuffer` and only returns on `io.EOF`, which deadlocks over a persistent TCP socket (the client is waiting for report-status). Hiding the capability falls through to `Parser.Parse`, which knows the pack's end from the format itself. HTTP doesn't need this — request bodies have a real EOF — so HTTP keeps the faster PackfileWriter path. Trade-off: git:// pushes write loose objects (one S3 PUT per object). 51 + 1. **`writePack` in `receivepack.go`** stores the incoming pack whole on every transport. go-git's default `PackfileWriter` path (`WritePackfileToObjectStorage` → `io.CopyBufferPool`) copies until `io.EOF`, which deadlocks over a persistent git:// / SSH socket (the client holds the connection open awaiting report-status). Instead of relying on EOF, `writePack` drives a `packfile.Scanner` over an `io.TeeReader(rd, packWriter)`: the scanner knows the pack's end from its own framing (header object count + trailer checksum) and stops there, while the tee mirrors exactly those bytes into the `PackfileWriter`. The result is one `.pack` + one `.idx` per push — no per-object loose writes (which on S3 cost a `HeadObject` dedup `Lstat` **and** a `PutObject` each). This replaced the old `streamingStorer` hack (which hid `PackfileWriter` to force the loose-object `Parser.Parse` path on git:// / SSH); all three transports now share `writePack`. 52 52 53 53 2. **No-op closers everywhere.** `transport.UploadPack`/`ReceivePack` call `Close` on the reader (and sometimes the writer) between negotiation rounds. The git:// socket can't survive that, and the HTTP `ResponseWriter` doesn't implement `Close`. Wrap with `io.NopCloser` (reader) and `ioutil.WriteNopCloser` from `go-git/v6/utils/ioutil` (writer). 54 54 55 - **SSH shares both gotchas with git://**, not HTTP: an `ssh.Session` is a persistent bidirectional stream, so `handleSSH` uses `streamingStorer{}` for receive-pack and wraps the session in the same no-op closers. (HTTP keeps the faster `PackfileWriter` path because the request body has a real EOF.) 55 + **SSH shares both gotchas with git://**, not HTTP: an `ssh.Session` is a persistent bidirectional stream, so `handleSSH` wraps the session in the same no-op closers and relies on the same Scanner-bounded `writePack` (gotcha #1) instead of the request-body EOF that HTTP enjoys. 56 + 57 + > **s3fs `ReadAt` contract.** Because packs are now read back via go-git's `packfile.FSObject` (which probes a packed object's handle with a 1-byte `ReadAt` and reopens the pack on `os.ErrClosed`), `s3fs`'s read file (`internal/s3fs/file.go`) returns `os.ErrClosed` from `Read`/`ReadAt`/`Seek` once closed rather than dereferencing its niled `*bytes.Reader`. A post-receive hook reading the just-pushed commit hits exactly this path. 56 58 57 59 ### The auth seam (`internal/auth`) 58 60

+1 -17

cmd/objgitd/git_protocol.go

··· 28 28 // It is cleared once the (possibly long) transfer begins. 29 29 const handshakeTimeout = 30 * time.Second 30 30 31 - // streamingStorer wraps a storage.Storer to hide its optional 32 - // storer.PackfileWriter capability. go-git's UpdateObjectStorage drains the 33 - // incoming pack into PackfileWriter via io.CopyBuffer, which only returns on 34 - // io.EOF — fine over HTTP (the request body has a natural EOF) but a deadlock 35 - // over git://, where the client holds the connection open waiting for the 36 - // server's report-status. With PackfileWriter hidden, UpdateObjectStorage 37 - // falls through to Parser.Parse, which knows the end of the pack from the 38 - // pack format itself and never waits for an EOF. 39 - // 40 - // Trade-off: Parser.Parse writes loose objects (one Rename → one S3 PUT each) 41 - // instead of one packfile, so large git:// pushes incur more S3 calls. HTTP 42 - // keeps the fast PackfileWriter path. 43 - type streamingStorer struct { 44 - storage.Storer 45 - } 46 - 47 31 // operationFor maps a git service to the access it needs: receive-pack writes, 48 32 // everything else (upload-pack, upload-archive) reads. 49 33 func operationFor(service string) auth.Operation { ··· 183 167 _, _ = pktline.WriteError(conn, fmt.Errorf("cannot open repository %q", req.Pathname)) 184 168 return fmt.Errorf("opening %q for push: %w", req.Pathname, err) 185 169 } 186 - return d.receivePack(ctx, streamingStorer{Storer: st}, st, req.Pathname, r, conn, &transport.ReceivePackRequest{ 170 + return d.receivePack(ctx, st, req.Pathname, r, conn, &transport.ReceivePackRequest{ 187 171 GitProtocol: gitProtocol, 188 172 }) 189 173

+86

cmd/objgitd/git_protocol_test.go

··· 10 10 "testing" 11 11 "time" 12 12 13 + "github.com/go-git/go-billy/v6" 13 14 "github.com/go-git/go-billy/v6/memfs" 14 15 "github.com/go-git/go-git/v6/plumbing/transport" 15 16 "tangled.org/xeiaso.net/objgit/internal/auth" ··· 111 112 if _, err := fs.Stat("/test.git/config"); err == nil { 112 113 t.Fatal("repository must not be created when push is disabled") 113 114 } 115 + } 116 + 117 + // TestDaemonPushKeepsPack verifies the receive-pack path stores the incoming 118 + // pack whole (objects/pack/pack-*.pack + .idx) rather than exploding it into 119 + // loose objects (objects/<2-hex>/...). git:// used to hide the PackfileWriter 120 + // capability and fall back to loose objects (one S3 PUT + one Lstat per object); 121 + // writePack now delimits the pack with a Scanner and feeds the PackfileWriter on 122 + // every transport. 123 + func TestDaemonPushKeepsPack(t *testing.T) { 124 + if _, err := exec.LookPath("git"); err != nil { 125 + t.Skip("git not installed") 126 + } 127 + 128 + fs := memfs.New() 129 + d := &daemon{ 130 + fs: fs, 131 + loader: transport.NewFilesystemLoader(fs, false), 132 + authz: auth.AllowAnonymous{AllowWrite: true}, 133 + } 134 + 135 + ctx, cancel := context.WithCancel(context.Background()) 136 + defer cancel() 137 + 138 + ln, err := net.Listen("tcp", "127.0.0.1:0") 139 + if err != nil { 140 + t.Fatalf("listen: %v", err) 141 + } 142 + go func() { _ = d.Serve(ctx, ln) }() 143 + 144 + remote := "git://" + ln.Addr().String() + "/test.git" 145 + 146 + work := t.TempDir() 147 + runGit(t, work, "init", "-b", "main") 148 + runGit(t, work, "config", "user.email", "test@example.com") 149 + runGit(t, work, "config", "user.name", "Test") 150 + writeFile(t, filepath.Join(work, "README.md"), "hello\n") 151 + runGit(t, work, "add", ".") 152 + runGit(t, work, "commit", "-m", "initial") // blob + tree + commit 153 + runGit(t, work, "push", remote, "main") 154 + 155 + assertPackedRepo(t, fs, "/test.git") 156 + } 157 + 158 + // assertPackedRepo fails unless repoPath holds at least one packfile and no loose 159 + // object directories (a 2-hex-char dir under objects/ such as "ab/"). 160 + func assertPackedRepo(t *testing.T, fs billy.Filesystem, repoPath string) { 161 + t.Helper() 162 + 163 + packs, err := fs.ReadDir(repoPath + "/objects/pack") 164 + if err != nil { 165 + t.Fatalf("ReadDir objects/pack: %v", err) 166 + } 167 + var packCount int 168 + for _, e := range packs { 169 + if strings.HasSuffix(e.Name(), ".pack") { 170 + packCount++ 171 + } 172 + } 173 + if packCount == 0 { 174 + t.Errorf("expected a packfile under %s/objects/pack, found none", repoPath) 175 + } 176 + 177 + entries, err := fs.ReadDir(repoPath + "/objects") 178 + if err != nil { 179 + t.Fatalf("ReadDir objects: %v", err) 180 + } 181 + for _, e := range entries { 182 + if e.IsDir() && isLooseObjectDir(e.Name()) { 183 + t.Errorf("found loose-object dir objects/%s; pack should have been kept whole", e.Name()) 184 + } 185 + } 186 + } 187 + 188 + // isLooseObjectDir reports whether name is a git loose-object fan-out directory 189 + // (two lowercase hex characters), distinguishing it from "pack" and "info". 190 + func isLooseObjectDir(name string) bool { 191 + if len(name) != 2 { 192 + return false 193 + } 194 + for _, c := range name { 195 + if !strings.ContainsRune("0123456789abcdef", c) { 196 + return false 197 + } 198 + } 199 + return true 114 200 } 115 201 116 202 func runGit(t *testing.T, dir string, args ...string) string {

+10 -9

cmd/objgitd/hooks.go

··· 81 81 // the repository's receive-pack hook for each updated branch once the push 82 82 // succeeds — synchronously, streaming hook output to the client over the 83 83 // sideband progress channel (rendered as "remote: " lines) before the response 84 - // stream is closed. rpStorer is what the service writes through (the git:// and 85 - // SSH paths hide the PackfileWriter capability via streamingStorer); readStorer 86 - // is the underlying storer used for ref snapshots and hook checkouts. 87 - func (d *daemon) receivePack(ctx context.Context, rpStorer, readStorer storage.Storer, repoPath string, r io.ReadCloser, w io.WriteCloser, req *transport.ReceivePackRequest) error { 84 + // stream is closed. st is both what the service writes through and the storer 85 + // used for ref snapshots and hook checkouts — all three transports now share the 86 + // same Scanner-bounded PackfileWriter path (see writePack), so no transport needs 87 + // a capability-hiding wrapper. 88 + func (d *daemon) receivePack(ctx context.Context, st storage.Storer, repoPath string, r io.ReadCloser, w io.WriteCloser, req *transport.ReceivePackRequest) error { 88 89 if !d.allowHooks { 89 - return receivePackStreaming(ctx, rpStorer, r, w, req, nil) 90 + return receivePackStreaming(ctx, st, r, w, req, nil) 90 91 } 91 92 92 - before, err := snapshotRefs(readStorer) 93 + before, err := snapshotRefs(st) 93 94 if err != nil { 94 95 slog.Warn("hook: ref snapshot before push failed", "path", repoPath, "err", err) 95 96 } ··· 99 100 // progress is the sideband band-2 writer, or nil when the client did not 100 101 // negotiate sideband (hooks then fall back to logging only). 101 102 onUpdated := func(progress io.Writer) { 102 - after, err := snapshotRefs(readStorer) 103 + after, err := snapshotRefs(st) 103 104 if err != nil { 104 105 slog.Error("hook: ref snapshot after push failed", "path", repoPath, "err", err) 105 106 return ··· 108 109 if len(updates) == 0 { 109 110 return 110 111 } 111 - d.runHooks(repoPath, "receive-pack", readStorer, updates, progress) 112 + d.runHooks(repoPath, "receive-pack", st, updates, progress) 112 113 } 113 114 114 - return receivePackStreaming(ctx, rpStorer, r, w, req, onUpdated) 115 + return receivePackStreaming(ctx, st, r, w, req, onUpdated) 115 116 } 116 117 117 118 // runHooks executes the receive-pack hook once per non-deleted branch update,

+1 -1

cmd/objgitd/http.go

··· 142 142 GitProtocol: gitProtocol, 143 143 }) 144 144 case transport.ReceivePackService: 145 - err = d.receivePack(r.Context(), st, st, repoPath, in, out, &transport.ReceivePackRequest{ 145 + err = d.receivePack(r.Context(), st, repoPath, in, out, &transport.ReceivePackRequest{ 146 146 StatelessRPC: true, 147 147 GitProtocol: gitProtocol, 148 148 })

+15 -44

cmd/objgitd/main.go

··· 45 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 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 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 48 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") 49 + s3CacheRecursive = flag.String("s3-cache-recursive-prefixes", "refs/", "comma-separated key prefixes served from one recursive subtree scan instead of a listing per folder; empty disables subtree caching") 50 + s3CacheMaxSubtree = flag.Int("s3-cache-max-subtree-keys", 50000, "abandon a recursive subtree scan past this many keys and fall back to per-folder listing") 53 51 ) 54 52 55 53 func main() { ··· 88 86 var cache *s3fs.ListingCache 89 87 var fsOpts []s3fs.Option 90 88 if *s3CacheTTL > 0 { 91 - var peers []string 92 - if *groupcachePeers != "" { 93 - peers = strings.Split(*groupcachePeers, ",") 89 + // Non-nil (even empty) so an empty flag explicitly disables subtree 90 + // caching rather than falling back to the {"refs/"} default. 91 + recursive := []string{} 92 + if *s3CacheRecursive != "" { 93 + recursive = strings.Split(*s3CacheRecursive, ",") 94 94 } 95 95 cache = s3fs.NewListingCache(s3fs.CacheConfig{ 96 - TTL: *s3CacheTTL, 97 - RefreshInterval: *s3CacheRefresh, 98 - IdleTTL: *s3CacheIdle, 99 - SizeBytes: *s3CacheSize, 100 - Self: *groupcacheSelf, 101 - Peers: peers, 96 + TTL: *s3CacheTTL, 97 + RefreshInterval: *s3CacheRefresh, 98 + IdleTTL: *s3CacheIdle, 99 + RecursivePrefixes: recursive, 100 + MaxSubtreeKeys: *s3CacheMaxSubtree, 102 101 }, client, *bucket, "/") 103 102 fsOpts = append(fsOpts, s3fs.WithListingCache(cache)) 104 103 metrics.RegisterListingCache(func() metrics.ListingCacheStats { 105 104 s := cache.Stats() 106 105 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, 106 + Hits: s.Hits, Misses: s.Misses, 107 + ListingItems: s.ListingItems, SubtreeItems: s.SubtreeItems, HeadItems: s.HeadItems, 111 108 } 112 109 }) 113 110 } ··· 136 133 "allow_hooks", *allowHooks, 137 134 "s3_cache_ttl", *s3CacheTTL, 138 135 "s3_cache_refresh", *s3CacheRefresh, 139 - "groupcache_self", *groupcacheSelf, 136 + "s3_cache_recursive_prefixes", *s3CacheRecursive, 140 137 ) 141 138 142 139 g, gCtx := errgroup.WithContext(ctx) ··· 146 143 cache.RunWarmer(gCtx) 147 144 return nil 148 145 }) 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 146 } 176 147 177 148 if *metricsBind != "" {

+41 -1

cmd/objgitd/receivepack.go

··· 6 6 "fmt" 7 7 "io" 8 8 9 + "github.com/go-git/go-git/v6/config" 9 10 "github.com/go-git/go-git/v6/plumbing" 11 + formatcfg "github.com/go-git/go-git/v6/plumbing/format/config" 10 12 "github.com/go-git/go-git/v6/plumbing/format/packfile" 11 13 "github.com/go-git/go-git/v6/plumbing/format/pktline" 12 14 "github.com/go-git/go-git/v6/plumbing/protocol" ··· 115 117 // Receive the packfile 116 118 var unpackErr error 117 119 if needPackfile { 118 - unpackErr = packfile.UpdateObjectStorage(st, rd) 120 + unpackErr = writePack(st, rd) 119 121 } 120 122 121 123 // Done with the request, now close the reader ··· 180 182 return firstErr 181 183 } 182 184 return closeWriter(w) 185 + } 186 + 187 + // writePack stores the incoming packfile as a single packfile object via the 188 + // storer's PackfileWriter, delimiting the pack with a Scanner rather than waiting 189 + // for the reader to reach io.EOF. go-git's default PackfileWriter path 190 + // (WritePackfileToObjectStorage → io.CopyBufferPool until EOF) deadlocks on a 191 + // persistent git:// / SSH socket, where the client holds the connection open 192 + // awaiting report-status. The Scanner knows the pack's end from its own framing 193 + // (header object count + trailer checksum) and stops there, while a TeeReader 194 + // mirrors exactly those bytes into the PackfileWriter — so the whole pack lands as 195 + // one object on every transport. Falls back to UpdateObjectStorage (loose objects) 196 + // if the storer cannot write packs. 197 + func writePack(st storage.Storer, rd io.Reader) error { 198 + pw, ok := st.(storer.PackfileWriter) 199 + if !ok { 200 + return packfile.UpdateObjectStorage(st, rd) 201 + } 202 + 203 + var sopts []packfile.ScannerOption 204 + if c, ok := st.(config.ConfigStorer); ok { 205 + if cfg, err := c.Config(); err == nil && cfg.Extensions.ObjectFormat == formatcfg.SHA256 { 206 + sopts = append(sopts, packfile.WithSHA256()) 207 + } 208 + } 209 + 210 + w, err := pw.PackfileWriter() 211 + if err != nil { 212 + return err 213 + } 214 + 215 + sc := packfile.NewScanner(io.TeeReader(rd, w), sopts...) 216 + for sc.Scan() { 217 + } 218 + if err := sc.Error(); err != nil { 219 + _ = w.Close() 220 + return err 221 + } 222 + return w.Close() 183 223 } 184 224 185 225 // sidebandProgress writes to the sideband progress channel (band 2), which the

+3 -3

cmd/objgitd/ssh.go

··· 219 219 _ = s.Exit(1) 220 220 return fmt.Errorf("opening %q for push: %w", repoPath, err) 221 221 } 222 - // streamingStorer hides PackfileWriter (the io.CopyBuffer-until-EOF path 223 - // deadlocks on a live socket); d.receivePack runs push hooks afterward. 224 - if err := d.receivePack(ctx, streamingStorer{Storer: st}, st, repoPath, r, w, &transport.ReceivePackRequest{}); err != nil { 222 + // d.receivePack stores the pack whole (Scanner-bounded PackfileWriter, see 223 + // writePack) and runs push hooks afterward. 224 + if err := d.receivePack(ctx, st, repoPath, r, w, &transport.ReceivePackRequest{}); err != nil { 225 225 slog.Error("ssh receive-pack failed", "path", repoPath, "err", err) 226 226 return err 227 227 }

+5

cmd/objgitd/ssh_test.go

··· 243 243 if !pushLanded && statErr == nil { 244 244 t.Fatal("repository must not exist when push did not land") 245 245 } 246 + if pushLanded { 247 + // SSH shares the Scanner-bounded PackfileWriter path: the push 248 + // must land as a packfile, not loose objects. 249 + assertPackedRepo(t, fs, "/test.git") 250 + } 246 251 247 252 dst := t.TempDir() 248 253 out, err := gitWithEnv(dst, env, "clone", remote, "cloned")

+211

docs/plans/git-keep-pushed-packs-whole.md

··· 1 + # Keep pushed packs whole on git:// and SSH 2 + 3 + ## Context 4 + 5 + Pushing big repositories to `objgitd` is slow. The root cause is not the go-git 6 + dotgit backend itself — it is that two of the three transports explode the 7 + received packfile into **loose objects**, one S3 object per git object. 8 + 9 + The git:// and SSH receive-pack paths wrap the storer in `streamingStorer{}` 10 + (`cmd/objgitd/git_protocol.go:43`) to **hide** the storer's `PackfileWriter` 11 + capability. That is a deliberate deadlock workaround: go-git's 12 + `WritePackfileToObjectStorage` copies the pack into the PackfileWriter with 13 + `io.CopyBufferPool`, which only returns on `io.EOF` — fine over HTTP (the request 14 + body has a real EOF) but a hang over a persistent git:// / SSH socket, where the 15 + client holds the connection open waiting for report-status. With PackfileWriter 16 + hidden, `packfile.UpdateObjectStorage` falls through to `Parser.Parse`, which 17 + knows the pack's end from its own framing and writes **loose objects**. 18 + 19 + Each loose object then costs **two** S3 round-trips in `internal/s3fs`: 20 + 21 + 1. go-git's `ObjectWriter.save()` (`dotgit/writers.go:375`) does 22 + `fs.Lstat(file)` before writing — a content-addressable dedup check — which 23 + becomes an S3 **HeadObject** (`s3fs/basic.go:200`), plus a ListObjectsV2 24 + directory probe on a miss. 25 + 2. The `Rename` of the temp object then becomes the **PutObject**. 26 + 27 + For a 100k-object push that is ~200k S3 round-trips. HTTP avoids all of this 28 + because it keeps the PackfileWriter path and stores the pack as a single object. 29 + 30 + **Goal:** make git:// and SSH keep the pack whole too — like HTTP, like real 31 + git's "keep large pack" behavior — so a push is ~2 S3 PutObjects (pack + idx) 32 + plus a handful of ref writes, instead of N. This also removes the 33 + stat-before-every-write entirely, because there are no per-object writes. 34 + 35 + ## Prerequisite (Step 0): fix the closed-pack-handle panic in s3fs 36 + 37 + This must land first — it is a latent crash that Approach B would otherwise newly 38 + expose on git:// and SSH (and which already crashes HTTP+hooks). 39 + 40 + go-git's `FSObject.Reader()` (`packfile/fsobject.go:62`) reads packed objects via 41 + `ReadAt` and explicitly recovers from a closed pack descriptor: it probes with a 42 + 1-byte `pack.ReadAt(...)` and, if the error matches `os.ErrClosed`, reopens the 43 + pack (`o.fs.Open(o.packPath)`). The billy contract is therefore that 44 + `File.ReadAt` on a closed file returns an `os.ErrClosed`-matching error. 45 + 46 + s3fs violates this: `s3ReadFile.Close()` sets `f.reader = nil` 47 + (`internal/s3fs/file.go:150`), but `Read`/`ReadAt`/`Seek` 48 + (`file.go:128/133/138`) dereference `f.reader` unconditionally, so a post-close 49 + call **panics on a nil `*bytes.Reader`** instead of returning `os.ErrClosed`. 50 + After a push the object LRU caches the new commit as an `FSObject` over a pack 51 + handle that index-building later closes; the post-receive hook's `GetCommit` then 52 + calls `Reader()` → probe `ReadAt` on the closed handle → panic 53 + (`hooks.go:136` in the reported trace). 54 + 55 + **Fix:** guard the three read methods of `s3ReadFile` to return `os.ErrClosed` 56 + when the file is closed, e.g.: 57 + 58 + ```go 59 + func (f *s3ReadFile) ReadAt(p []byte, off int64) (int, error) { 60 + if f.closed || f.reader == nil { 61 + return 0, os.ErrClosed 62 + } 63 + return f.reader.ReadAt(p, off) 64 + } 65 + ``` 66 + 67 + Apply the same guard to `Read` and `Seek`. (`os` is already imported in 68 + `file.go`.) Return `os.ErrClosed` specifically — not the existing `ErrFileClosed` 69 + — unless `ErrFileClosed` is made to wrap it, because go-git keys its reopen on 70 + `errors.Is(err, os.ErrClosed)`. 71 + 72 + **Caveat / follow-up (out of scope):** s3fs `Open` reads the _whole_ object into 73 + memory (`file.go:95`), so go-git's reopen-on-closed downloads the entire pack 74 + again per cache-resident `FSObject` read. Acceptable for hooks (a handful of 75 + reads), but a future optimization is to serve `ReadAt` via S3 `GetObject` with a 76 + `Range` header (true random access) and/or keep pack descriptors open, instead of 77 + buffering the full pack. Note it; don't build it here. 78 + 79 + ## Approach: drive PackfileWriter with a Scanner-bounded TeeReader 80 + 81 + The deadlock only exists because `io.CopyBufferPool` waits for the reader's EOF. 82 + We don't need EOF: the packfile is self-delimiting (header carries the object 83 + count; a trailer checksum ends it). go-git's `packfile.Scanner` 84 + (`plumbing/format/packfile/scanner.go`) walks exactly those bytes and stops at 85 + the trailer — this is the same machinery `Parser.Parse` already uses successfully 86 + over the socket today, so it provably terminates without blocking (the client 87 + sends nothing after the pack, so the underlying socket read never waits past the 88 + trailer). 89 + 90 + So we get the PackfileWriter's writer, tee the socket reader through it, and let 91 + the Scanner consume exactly the pack — `w.Close()` then builds the index and 92 + uploads one `.pack` + one `.idx`. This depends on framing, not EOF, so it works 93 + identically on HTTP, git://, and SSH. The `streamingStorer` workaround is no 94 + longer needed and is deleted. 95 + 96 + ### Changes 97 + 98 + **1. `cmd/objgitd/receivepack.go` — add `writePack`, replace the unpack call.** 99 + 100 + Replace the body at line 118 (`unpackErr = packfile.UpdateObjectStorage(st, rd)`) 101 + with a call to a new helper: 102 + 103 + ```go 104 + // writePack stores the incoming packfile as a single packfile object via the 105 + // storer's PackfileWriter, delimiting the pack with a Scanner rather than 106 + // waiting for the reader to reach io.EOF. go-git's default PackfileWriter path 107 + // (io.CopyBufferPool until EOF) deadlocks on a persistent git:// / SSH socket 108 + // where the client holds the connection open awaiting report-status; the 109 + // Scanner knows the pack's end from its own framing and stops there, while a 110 + // TeeReader mirrors exactly those bytes into the PackfileWriter. Falls back to 111 + // UpdateObjectStorage (loose objects) if the storer cannot write packs. 112 + func writePack(st storage.Storer, rd io.Reader) error { 113 + pw, ok := st.(storer.PackfileWriter) 114 + if !ok { 115 + return packfile.UpdateObjectStorage(st, rd) 116 + } 117 + 118 + var sopts []packfile.ScannerOption 119 + if c, ok := st.(config.ConfigStorer); ok { 120 + if cfg, err := c.Config(); err == nil && 121 + cfg.Extensions.ObjectFormat == formatcfg.SHA256 { 122 + sopts = append(sopts, packfile.WithSHA256()) 123 + } 124 + } 125 + 126 + w, err := pw.PackfileWriter() 127 + if err != nil { 128 + return err 129 + } 130 + sc := packfile.NewScanner(io.TeeReader(rd, w), sopts...) 131 + for sc.Scan() { 132 + } 133 + if err := sc.Error(); err != nil { 134 + _ = w.Close() 135 + return err 136 + } 137 + return w.Close() 138 + } 139 + ``` 140 + 141 + - Object-format detection mirrors `packfile.UpdateObjectStorage` 142 + (`common.go:29`): read it from the `ConfigStorer`, default SHA-1; the only 143 + scanner knob is `packfile.WithSHA256()` (`scanner_options.go:16`). 144 + - New imports: `config`, `formatcfg "…/plumbing/format/config"`, 145 + and `storer` is already imported. (`packfile`, `storage`, `io` already present.) 146 + - Call site becomes `unpackErr = writePack(st, rd)`. 147 + 148 + **2. `cmd/objgitd/git_protocol.go` — delete `streamingStorer`.** 149 + 150 + Remove the type and its doc comment (lines 31–45). The `storage` import may 151 + become unused here — drop it if so. 152 + 153 + **3. Collapse `receivePack`'s two storer params to one (`hooks.go` + call sites).** 154 + 155 + `streamingStorer` was the _only_ reason `rpStorer` and `readStorer` differed; all 156 + callers will now pass the same `st` for both. Simplify 157 + `(*daemon).receivePack` (`hooks.go:87`) to a single `st storage.Storer` 158 + param (used for both the service write and the ref snapshots/hook checkout), 159 + update its doc comment, and update the three call sites: 160 + 161 + - `git_protocol.go:186` — `d.receivePack(ctx, st, req.Pathname, …)` 162 + - `ssh.go:224` — `d.receivePack(ctx, st, repoPath, …)`; delete the now-stale 163 + `streamingStorer` comment at `ssh.go:222–223`. 164 + - `http.go:145` — `d.receivePack(r.Context(), st, repoPath, …)` 165 + 166 + **4. Update docs.** `CLAUDE.md` "Two subtle protocol points" #1 and the SSH 167 + section both describe the `streamingStorer` workaround and the loose-object 168 + trade-off — rewrite to describe the Scanner-bounded PackfileWriter that all three 169 + transports now share. (`docs/plans/` may also get a short note.) 170 + 171 + ### Edge cases / risk 172 + 173 + - **No over-read / no hang:** the Scanner's internal `bufio` only issues a read 174 + when its buffer drains, and a socket read returns the bytes currently present; 175 + the client sends nothing after the trailer, so it never blocks past the pack. 176 + This is the exact behavior the current `Parser.Parse` path relies on. 177 + - **Memory:** the pack is buffered in the s3fs in-memory `tempBuffer` before 178 + upload (`internal/s3fs/tempfs.go`) — identical to today's HTTP path, so git:// 179 + / SSH now match HTTP's memory profile. Spilling large packs to disk / multipart 180 + upload is a possible follow-up, out of scope here. 181 + - **Empty/delete-only pushes:** unchanged — `needPackfile` 182 + (`receivepack.go:108`) already gates `writePack` to pushes that carry a pack. 183 + - **Delta bases in existing objects:** unaffected — reads still go through the 184 + normal storer; we only changed how the incoming pack is _written_. 185 + 186 + ## Verification 187 + 188 + 1. `go build ./...` 189 + 2. `go test ./...` — existing round-trip push/clone tests for all three 190 + transports must stay green (`git_protocol_test.go`, `ssh_test.go`, 191 + `http_test.go`; gated on `git`/`ssh` in PATH). 192 + 3. **New assertion proving packs are kept** (tests back the daemon with 193 + `memfs.New()`, so the stored layout is directly inspectable): after a git:// 194 + push and an SSH push, assert the repo contains `objects/pack/pack-*.pack` 195 + (+`.idx`) and that **no** loose-object dirs (`objects/<2-hex>/`) exist. Add as 196 + a table case alongside the existing receive-pack tests; reuse `seedRepo` / 197 + `runGit` helpers from `git_protocol_test.go`. 198 + 4. **Regression test for the Step 0 panic:** with `allowHooks: true` and a repo 199 + whose pushed tree contains `.objgit/hooks/receive-pack`, push over **HTTP** 200 + (today's crash) and over **git:// / SSH** (newly pack-backed after Approach B) 201 + and assert the push succeeds and the hook runs — exercising 202 + `GetCommit` → `FSObject.Reader()` → `ReadAt` on a cache-resident, closed pack 203 + handle without panicking. A focused s3fs unit test also helps: open a file, 204 + `Close()` it, then assert `ReadAt`/`Read`/`Seek` each return an 205 + `errors.Is(_, os.ErrClosed)` error rather than panicking. 206 + 5. **Manual, against a real bucket:** push a large repo over SSH 207 + (`./objgitd -bucket $BUCKET -ssh-bind :2222 -allow-push`) and watch 208 + `objgit_s3_requests_total{op="PutObject"}` / `{op="HeadObject"}` on `/metrics` 209 + — PutObject should drop from O(objects) to ~2 (pack + idx) and HeadObject 210 + should no longer scale with object count. Compare wall-clock against a 211 + pre-change push of the same repo.

+168 -131

docs/plans/s3fs-groupcache-listobjectsv2.md

··· 1 - # Plan: groupcache-backed directory-listing cache for `internal/s3fs` 1 + # Plan: directory-listing cache for `internal/s3fs` 2 + 3 + > **Status note.** This started as a groupcache-backed, fleet-shareable cache 4 + > (hence the filename). It shipped as a **process-local `sync.Map` cache** — 5 + > groupcache was ripped out for simplicity. There is no cross-process sharing, 6 + > no peer pool, and no window-encoded key; TTL is a plain per-entry expiry. The 7 + > sections below are revised to match what shipped; ignore any lingering peer/ 8 + > window phrasing. 2 9 3 10 ## Context 4 11 ··· 17 24 `Stat`/`Open` touching an un-cached folder lists that **parent folder** in full and 18 25 populates the cache, warming every sibling at once. 19 26 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**: 27 + **Backend: process-local `sync.Map`s** (one for folder listings, one for recursive 28 + subtrees, one for per-object heads). Each cached entry carries two pieces of bookkeeping: 24 29 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. 30 + - **Per-entry TTL.** An entry stores `expires = now + TTL`; past it the entry is ignored 31 + and re-listed. This bounds how long a write this process can't see stays hidden. (There 32 + is no background warmer requirement for correctness; the warmer just keeps hot entries 33 + fresh and sweeps expired ones, since a `sync.Map` has no LRU.) 34 + - **Per-prefix local generation for precise invalidation.** A process-local counter per 35 + prefix is bumped on every local write under that prefix (and its ancestors — see 36 + subtree caching). An entry whose stored generation no longer matches is ignored, so the 37 + next read re-lists and sees the write immediately (read-after-write correctness). 38 + Concurrent identical fills are coalesced with `golang.org/x/sync/singleflight`. 34 39 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. 40 + **Accepted limitation:** negative staleness is bounded by the TTL — a just-deleted-or- 41 + created object may read stale for up to one TTL. Safe for git's _content_ (objects are 42 + immutable and content-addressed, so positive listings never go wrong); the TTL bounds the 43 + negative-staleness risk. Operators tune it with `-s3-cache-ttl`; `-s3-cache-ttl 0` 44 + disables the cache and restores today's exact behavior. 42 45 43 - **Defaults:** on by default, tunable; single-process unless peers are configured. 46 + **Defaults:** on by default, tunable; always single-process. 44 47 45 48 ## Correctness model 46 49 ··· 48 51 with a different `root`, so the `*ListingCache` is **shared by pointer** across a root 49 52 fs and all chroot children, and prefixes are full-canonical (`fs3.key()`/`cleanPath` — 50 53 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. 54 + - **Cache key** is just the canonical prefix; the entry holds `{gen, expires}` alongside 55 + its payload. A read hits only when `entry.gen == gen(prefix)` and `now < entry.expires`. 56 + - **Both `ReadDir` and `Stat`/`Open` route through `list()`**, so each folder is listed 57 + once and the result feeds both; `singleflight` coalesces concurrent identical fills (one 58 + list even when a clone fans out across siblings). 59 59 - **Listing payload** carries enough to serve both consumers without a second call: 60 60 per child `{name, kind(file|dir), size, mtimeUnixNano}` (`CommonPrefixes`→dir with 61 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). 62 + matching today's Head-first precedence). Stored as a Go value in the `sync.Map` — no 63 + serialization. 64 64 - **In-process writes bump the local generation** of the parent prefix at _write 65 65 completion_, moving the key so the next read re-lists: 66 66 - `s3WriteFile.Close` / `s3MultipartUploadFile.Close` — after the upload succeeds. 67 67 - `Rename` (covers `TempFile`→final pack promotion), `Remove`, `MkdirAll`. 68 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`. 69 + cache** (see [Head cache](#head-cache)), rather than a foreground `HeadObject`. The head 70 + cache is **seeded straight from each listing** — `ListObjectsV2` already returns every 71 + file's size and mtime — so a positive lookup costs no extra round-trip. Listings omit 72 + the user-metadata the unix-metadata feature needs, so a caller that requires it (i.e. 73 + unix-metadata enabled) treats a listing-seeded entry as a miss and fills via a real 74 + `HeadObject`. `Open` still issues `GetObject` for the body but skips its `HeadObject`. A 75 + `NoSuchKey` from a delete racing the listing maps to `NotExist`. 74 76 75 77 ## Changes 76 78 77 79 ### go.mod 78 80 79 - - Add `github.com/golang/groupcache` (effectively dependency-free). 81 + - No new direct dependency. `golang.org/x/sync/singleflight` (already vendored for 82 + `errgroup`) dedupes concurrent fills. 80 83 81 84 ### New file: `internal/s3fs/listingcache.go` 82 85 83 - The cache wrapper, groupcache group, key logic, and background warmer. 86 + The three `sync.Map` caches, the TTL/generation key logic, and the background warmer. 84 87 85 88 ```go 86 89 type CacheConfig struct { 87 90 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 91 + DisableHeadPrefetch bool // zero value = seed heads from listings on 92 + RecursivePrefixes []string // nil → {"refs/"}; empty → subtree caching off 93 + MaxSubtreeKeys int // <=0 → 50000 93 94 } 94 95 95 96 type childKind uint8 // kindFile / kindDir 96 97 type childEntry struct { Name string; Kind childKind; Size, Mtime int64 } 97 98 type headData struct { Size, Mtime int64; Meta map[string]string } 99 + type headCacheEntry struct { data headData; gen uint64; expires time.Time; hasMeta bool } 100 + type listingEntry struct { entries []childEntry; gen uint64; expires time.Time } 101 + type subtreeEntry struct { data subtreeData; gen uint64; expires time.Time } 98 102 99 103 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 104 + ttl time.Duration 105 + cfg CacheConfig 106 + client s3Client 107 + bucket string 108 + separator string 109 + roots []string // normalised RecursivePrefixes, longest first 110 + clock func() time.Time // overridable in tests 111 + listings sync.Map // prefix → listingEntry 112 + subtrees sync.Map // root → subtreeEntry 113 + heads sync.Map // object key → headCacheEntry 114 + sf, headSF singleflight.Group // coalesce concurrent fills 115 + hits, misses atomic.Int64 // for metrics 116 + mu sync.Mutex 117 + gens map[string]uint64 // per-prefix local generation 118 + seen map[string]time.Time // prefixes accessed → driven by the warmer 110 119 } 111 120 ``` 112 121 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. 122 + - `NewListingCache(cfg, client, bucket, separator)` — applies defaults and normalises the 123 + recursive roots; no groups, no pool. 124 + - `list(ctx, prefix) ([]childEntry, error)` — the one entry point both `ReadDir` and 125 + `Stat`/`Open` use. Routes recursive prefixes to `subtree`; otherwise `listFolder` 126 + (sync.Map lookup gated on gen+expiry, `fillFolder` via singleflight on a miss). 127 + - `gen(prefix)` / `invalidate(prefix)` — `invalidate` bumps the generation of `prefix` 128 + **and every ancestor** and records `seen` (so the warmer refreshes them). 129 + - `RunWarmer(ctx)` — `time.NewTicker(RefreshInterval)`; each tick drops `seen` entries idle 130 + past `IdleTTL`, re-fills the rest (routing recursive→subtree, deduped), then sweeps 131 + expired entries from all three maps (no LRU, so the warmer bounds growth). No-op when 132 + `RefreshInterval<=0`. Returns on `ctx.Done()`. 133 + - `Stats()` accessor exports hit/miss counters and resident item counts to metrics. 129 134 - `splitKey(key) (prefix, base)` — split on last `/`; no slash → `("", key)`. 130 135 131 - ### Head precache 136 + ### Head cache 137 + 138 + The per-object head cache is a **`sync.Map`** of `headCacheEntry`, keyed by 139 + canonical object key — like the listing and subtree caches. An 140 + entry is a hit only while (a) unexpired (`expires`, one TTL past fill), (b) still tagged 141 + with its parent prefix's current local generation — so `invalidate`'s generation bump 142 + drops every cached head under the prefix without a map scan, mirroring listing 143 + invalidation — and (c) carrying user metadata when the caller needs it (`hasMeta`). 132 144 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. 145 + - **Seeded from listings (`seedHeads`), not separately fetched.** When the listing getter 146 + fills a folder, it stores a `headCacheEntry` for every file **directly from the 147 + `ListObjectsV2` data** (size + mtime), with `hasMeta=false`. This warms the head cache 148 + with **zero** extra `HeadObject` calls — the listing already paid for the data. 149 + `CacheConfig.DisableHeadPrefetch` turns seeding off (zero value = on). 150 + - **`headInfo(ctx, key, needMeta) (*s3.HeadObjectOutput, error)`** serves a foreground 151 + lookup. A warm hit costs no round-trip. `needMeta` reports whether the caller needs the 152 + x-amz-meta-* user metadata (true iff unix-metadata is enabled): when true, a 153 + listing-seeded (`hasMeta=false`) entry is treated as a miss and filled via one real 154 + `HeadObject` (which stores `hasMeta=true`). `headSF` (a `singleflight.Group`) dedupes 155 + concurrent fills for the same key. 156 + - The warmer also **sweeps expired head entries** each tick — the `sync.Map` has no LRU, 157 + so the warmer is what bounds its growth to roughly the live working set. 158 + - `Stat`/`Open` of a present file call `headInfo` with `needMeta = fs3.unixMeta != nil`; 159 + `newS3ReadFile` gains an optional precomputed `*s3.HeadObjectOutput` so it skips its own 160 + `HeadObject` and only `GetObject`s the body. 138 161 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. 162 + ### Subtree caching (`RecursivePrefixes`) 163 + 164 + Caching one folder per `ListObjectsV2` is wasteful for bounded namespaces that callers 165 + walk folder-by-folder — `refs/` above all (`refs/`, `refs/heads/`, `refs/tags/`, 166 + `refs/remotes/…`, each its own delimited list). A **single delimiter-less 167 + `ListObjectsV2` over `refs/`** returns the whole subtree; every descendant folder's 168 + listing and every negative lookup beneath it is then synthesised in memory. 169 + 170 + - **A second `sync.Map`** (`subtrees`, keyed by root) holds `subtreeData{ Objects 171 + []subtreeObject; Truncated bool }` — the flat key+size+mtime set under a root — in a 172 + `subtreeEntry{gen, expires}`, the same TTL/generation scheme as folder listings. 173 + - **`list(prefix)` routes**: if `recursiveRoot(prefix)` matches a configured root, serve 174 + from `subtree(root)` via `synthesizeListing(objects, prefix)` (remainder-after-prefix 175 + with a `/` ⇒ child dir, deduped; else child file). Otherwise the existing delimited 176 + path. A complete subtree is authoritative for negative lookups (it has *all* keys under 177 + the root); only `!Truncated` subtrees are trusted. 178 + - **Bounded.** `listSubtree` stops once it exceeds `MaxSubtreeKeys` (default 50000) and 179 + reports `Truncated`; `list` then **falls back to the delimited per-folder listing**, so 180 + an unbounded namespace can't blow up memory. The truncated marker is itself cached, so 181 + the fallback costs one near-free subtree-cache hit plus the folder list. 182 + - **Invalidation walks ancestors.** `invalidate(prefix)` now bumps the generation of 183 + `prefix` **and every parent up to `""`** (`ancestorPrefixes`), so a write to 184 + `refs/tags/v1` moves the `refs/` subtree key (and the root listing's). Trade-off: a 185 + broader blast radius — a write also re-lists the coarser folders above it on their next 186 + read. Acceptable because writes are pushes and a re-list after a push is expected. 187 + - **Head seeding** extends to subtrees: a complete scan seeds every file's head 188 + (`seedSubtreeHeads`), each tagged with its own parent prefix's generation. A truncated 189 + scan seeds nothing (leaves heads to the fallback path). 190 + - **Warmer routing** mirrors `list`: seen prefixes under a recursive root warm that root's 191 + subtree (deduped across siblings) rather than per-folder. 192 + - **Config**: `CacheConfig.RecursivePrefixes` (nil ⇒ `{"refs/"}`; explicit empty ⇒ off) 193 + and `MaxSubtreeKeys`. `main.go` exposes `-s3-cache-recursive-prefixes` (default `refs/`, 194 + empty disables) and `-s3-cache-max-subtree-keys` (default 50000). 151 195 152 196 ### `internal/s3fs/filesystem.go` 153 197 ··· 164 208 - Extract `listChildren(ctx, client, bucket, separator, prefix) ([]childEntry, error)` — 165 209 the paginated `ListObjectsV2` loop, classifying `CommonPrefixes`→dir, `Contents`→file 166 210 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. 211 + (which holds only the raw client) can reuse it. Used by `ReadDir` (cache off) and the 212 + listing getter. 169 213 - `ReadDir`: when `cache != nil`, get the entries via `cache.list(ctx, prefix)` and 170 214 build `[]fs.DirEntry` from them (rebuilding `newDirInfo`/`newFileInfo` from the payload); 171 215 otherwise list directly as today. ··· 196 240 197 241 ### `internal/metrics/metrics.go` 198 242 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. 243 + - A Prometheus collector that reads `ListingCache.Stats()` (`{Hits, Misses, ListingItems, 244 + SubtreeItems, HeadItems}`) and exports `objgit_s3_listing_cache_hits_total`, 245 + `_misses_total`, and `_items{kind=listing|subtree|head}`. No `repo` label. (Cache fills 246 + are already counted as `ListObjectsV2`/`HeadObject` via `observeS3`.) `main` registers it 247 + only when the cache is enabled. 204 248 205 249 ### `cmd/objgitd/main.go` 206 250 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. 251 + - Flags (kebab-case + flagenv): 252 + - `-s3-cache-ttl` (`Duration`, default `60s`) — per-entry TTL; `<=0` disables the cache. 209 253 - `-s3-cache-refresh` (`Duration`, default `30s`) — warmer interval; `<=0` disables the 210 254 warmer (lazy fill still works). 211 255 - `-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()`. 256 + - `-s3-cache-recursive-prefixes` (`String`, default `refs/`) — comma-separated subtree 257 + roots; empty disables subtree caching. 258 + - `-s3-cache-max-subtree-keys` (`Int`, default `50000`) — subtree scan cap. 217 259 - 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 })`. 260 + `s3fs.WithListingCache(cache)` into `NewS3FS`, register the metrics collector, and add 261 + `g.Go(func() error { cache.RunWarmer(gCtx); return nil })` to the errgroup. 223 262 - Add the cache settings to the startup `slog.Info` line. 224 263 225 264 ## Why this is acceptably safe 226 265 227 266 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. 267 + so even a 60s TTL eliminates essentially all redundant `HeadObject`+probe pairs, and the 268 + first miss in a folder warms every sibling via one parent listing (deduped by 269 + singleflight). Local writes bump the per-prefix generation (and its ancestors'), so a 270 + push reads its own objects immediately. git object _content_ is immutable, so 271 + positive/`ReadDir` results are never wrong about what exists — only the _recency_ of 272 + newly-added entries is TTL-bounded. The residual risk is a negative read of an object 273 + another process just created (this cache is process-local), bounded by one TTL; `-s3-cache-ttl 0` 274 + opts out entirely. 240 275 241 276 ## Verification 242 277 243 278 1. `go build ./...`; `go mod tidy`; `go test ./...`. 244 279 2. **Cache disabled = no behavior change:** the cache is only wired when `-s3-cache-ttl>0`; 245 280 existing protocol tests (`go test ./cmd/objgitd/...`, needs `git` on PATH) must pass 246 - with the cache off and on (single-process, no peers). 281 + with the cache off and on. 247 282 3. **New unit tests in `internal/s3fs`** (table-driven `tt`, counting-stub `storage.Client`): 248 283 - **Populate-on-miss:** first `Stat` of an absent key in a never-listed folder issues 249 284 exactly one `ListObjectsV2` (the parent) and **zero** `HeadObject`; a second absent ··· 251 286 - `ReadDir` then `Stat`/`Open` of an absent sibling → zero S3; a dir child → zero S3. 252 287 - **Local invalidation:** `Create`+`Close` / `Rename` / `Remove` / `MkdirAll` bump the 253 288 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`. 289 + - **TTL expiry:** advancing time past `TTL` expires the entry → re-list (inject a clock 290 + or a settable `now` in `ListingCache` for the test). 291 + - **Warmer:** `RunWarmer` re-fills accessed prefixes and evicts idle ones past `IdleTTL`. 257 292 - **Chroot sharing:** `ReadDir` on the root then `Stat` of an absent child through a 258 293 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 294 + - **Head seeding:** one listing fill seeds every file's head from the `ListObjectsV2` 295 + data with **zero** `HeadObject`s; `Stat` of a seeded file then does **zero** further 296 + `HeadObject`s (counting-stub tests disable seeding for determinism; a dedicated test 297 + enables it and asserts no heads are issued). 298 + - **Subtree caching:** one read of any `refs/` folder scans the subtree once; other 299 + `refs/` folders and negative lookups beneath them then do **zero** S3; a write to a 300 + sibling `refs/` folder re-scans (ancestor invalidation) and is visible; a subtree 301 + past `MaxSubtreeKeys` falls back to a delimited listing yet still returns correctly. 302 + 4. **End-to-end:** `./objgitd -bucket $BUCKET -allow-push`; clone a packed repo twice and 266 303 confirm `objgit_s3_requests_total{operation="HeadObject"}` grows far slower than with 267 - `-s3-cache-ttl 0`. 304 + `-s3-cache-ttl 0`, and watch `objgit_s3_listing_cache_hits_total` climb.

-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 14 13 github.com/joho/godotenv v1.5.1 15 14 github.com/prometheus/client_golang v1.23.2 16 15 github.com/tigrisdata/storage-go v0.6.0 ··· 48 47 github.com/facebookgo/stack v0.0.0-20160209184415-751773369052 // indirect 49 48 github.com/facebookgo/subset v0.0.0-20200203212716-c811ad88dec4 // indirect 50 49 github.com/go-git/gcfg/v2 v2.0.2 // indirect 51 - github.com/golang/protobuf v1.5.4 // indirect 52 50 github.com/kevinburke/ssh_config v1.6.0 // indirect 53 51 github.com/klauspost/cpuid/v2 v2.3.0 // indirect 54 52 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect

-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= 82 78 github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= 83 79 github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= 84 80 github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0=

+15 -35

internal/metrics/metrics.go

··· 157 157 } 158 158 159 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. 160 + // counters. It mirrors s3fs.CacheStats but is defined here so s3fs stays free of 161 + // any Prometheus import; main bridges the two. 162 162 type ListingCacheStats struct { 163 - Gets, CacheHits, Loads, LocalLoads, PeerLoads, LocalLoadErrs int64 164 - MainBytes, MainItems, MainEvictions int64 165 - HotBytes, HotItems int64 163 + Hits, Misses int64 164 + ListingItems, SubtreeItems, HeadItems int64 166 165 } 167 166 168 167 // 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. 168 + // directory-listing cache's counters under objgit_s3_listing_cache_*. provider 169 + // is polled at scrape time. Call once at startup when the cache is enabled. 172 170 func RegisterListingCache(provider func() ListingCacheStats) { 173 171 prometheus.MustRegister(&listingCacheCollector{provider: provider}) 174 172 } ··· 178 176 } 179 177 180 178 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) 179 + lcHits = prometheus.NewDesc("objgit_s3_listing_cache_hits_total", "Listing/subtree-cache lookups served from cache.", nil, nil) 180 + lcMisses = prometheus.NewDesc("objgit_s3_listing_cache_misses_total", "Listing/subtree-cache lookups that fell through to S3.", nil, nil) 181 + lcItems = prometheus.NewDesc("objgit_s3_listing_cache_items", "Resident cache entries by kind.", []string{"kind"}, nil) 190 182 ) 191 183 192 184 func (c *listingCacheCollector) Describe(ch chan<- *prometheus.Desc) { 193 - ch <- lcGets 194 185 ch <- lcHits 195 - ch <- lcLoads 196 - ch <- lcLocal 197 - ch <- lcPeer 198 - ch <- lcLoadErrs 199 - ch <- lcBytes 186 + ch <- lcMisses 200 187 ch <- lcItems 201 - ch <- lcEvictions 202 188 } 203 189 204 190 func (c *listingCacheCollector) Collect(ch chan<- prometheus.Metric) { 205 191 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") 192 + ch <- prometheus.MustNewConstMetric(lcHits, prometheus.CounterValue, float64(s.Hits)) 193 + ch <- prometheus.MustNewConstMetric(lcMisses, prometheus.CounterValue, float64(s.Misses)) 194 + ch <- prometheus.MustNewConstMetric(lcItems, prometheus.GaugeValue, float64(s.ListingItems), "listing") 195 + ch <- prometheus.MustNewConstMetric(lcItems, prometheus.GaugeValue, float64(s.SubtreeItems), "subtree") 196 + ch <- prometheus.MustNewConstMetric(lcItems, prometheus.GaugeValue, float64(s.HeadItems), "head") 217 197 }

+4 -4

internal/s3fs/basic.go

··· 76 76 77 77 // If the parent folder's listing is cached, resolve the open without a 78 78 // negotiation round-trip: absent → not-exist, a sub-prefix → directory. 79 - // For a present file the head cache (prewarmed in the background) lets 79 + // For a present file the head cache (seeded from the listing) lets 80 80 // newS3ReadFile skip its HeadObject; only GetObject fetches the body. 81 81 if info, found, known := fs3.resolve(context.TODO(), key); known { 82 82 switch { ··· 85 85 case info.Kind == kindDir: 86 86 return newS3DirFile(key, fs3.bucket, fs3.client), nil 87 87 default: 88 - if ho, err := fs3.cache.headInfo(context.TODO(), key); err == nil { 88 + if ho, err := fs3.cache.headInfo(context.TODO(), key, fs3.unixMeta != nil); err == nil { 89 89 return newS3ReadFile(fs3.client, fs3.bucket, key, filename, ho) 90 90 } else if isNotFound(err) { 91 91 return nil, &os.PathError{Op: "open", Path: filename, Err: fs.ErrNotExist} ··· 180 180 181 181 // If the parent folder's listing is cached, answer without a foreground S3 182 182 // round-trip: absent → not-exist, a sub-prefix → directory, a present file 183 - // → the head cache (prewarmed in the background from the listing). 183 + // → the head cache (seeded from the listing's size/mtime). 184 184 if info, found, known := fs3.resolve(ctx, key); known { 185 185 if !found { 186 186 return nil, &os.PathError{Op: "stat", Path: filename, Err: fs.ErrNotExist} ··· 188 188 if info.Kind == kindDir { 189 189 return newDirInfo(path.Base(key)), nil 190 190 } 191 - if ho, err := fs3.cache.headInfo(ctx, key); err == nil { 191 + if ho, err := fs3.cache.headInfo(ctx, key, fs3.unixMeta != nil); err == nil { 192 192 return newFileInfoFromHead(path.Base(key), ho, fs3.unixMeta), nil 193 193 } else if isNotFound(err) { 194 194 return nil, &os.PathError{Op: "stat", Path: filename, Err: fs.ErrNotExist}

+38

internal/s3fs/dir.go

··· 72 72 return append(dirs, files...), nil 73 73 } 74 74 75 + // listSubtree lists every object under root with a single delimiter-less 76 + // (recursive) paginated ListObjectsV2, returning each object's full key plus 77 + // size/mtime. It stops once more than maxKeys objects have accumulated and 78 + // reports truncated=true; the caller then abandons the subtree and falls back to 79 + // delimited per-folder listing, so an unbounded namespace can't blow up memory. 80 + func listSubtree(ctx context.Context, client s3Client, bucket, root string, maxKeys int) (objs []subtreeObject, truncated bool, err error) { 81 + var ct *string 82 + for { 83 + start := time.Now() 84 + res, lerr := client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{ 85 + Bucket: &bucket, 86 + Prefix: &root, 87 + ContinuationToken: ct, 88 + // No Delimiter: a recursive listing of the whole subtree. 89 + }) 90 + observeS3("ListObjectsV2", start, lerr) 91 + if lerr != nil { 92 + return nil, false, lerr 93 + } 94 + 95 + for _, o := range res.Contents { 96 + objs = append(objs, subtreeObject{ 97 + Key: aws.ToString(o.Key), 98 + Size: aws.ToInt64(o.Size), 99 + Mtime: aws.ToTime(o.LastModified).UnixNano(), 100 + }) 101 + } 102 + 103 + if maxKeys > 0 && len(objs) > maxKeys { 104 + return objs, true, nil 105 + } 106 + if !aws.ToBool(res.IsTruncated) { 107 + return objs, false, nil 108 + } 109 + ct = res.NextContinuationToken 110 + } 111 + } 112 + 75 113 // ReadDir reads the directory named by dirname and returns a list of 76 114 // directory entries. When a listing cache is attached the listing is served 77 115 // through it (and reused by later Stat/Open of siblings).

+14 -1

internal/s3fs/file.go

··· 126 126 127 127 // Read implements io.Reader for billy.File 128 128 func (f *s3ReadFile) Read(p []byte) (n int, err error) { 129 + if f.closed || f.reader == nil { 130 + return 0, os.ErrClosed 131 + } 129 132 return f.reader.Read(p) 130 133 } 131 134 132 - // ReadAt implements io.ReaderAt for billy.File 135 + // ReadAt implements io.ReaderAt for billy.File. It returns os.ErrClosed once the 136 + // file is closed (rather than dereferencing the nil reader and panicking): go-git's 137 + // packfile.FSObject.Reader probes with a 1-byte ReadAt and reopens the pack when it 138 + // sees an os.ErrClosed-matching error, so a cache-resident object over a closed pack 139 + // handle recovers instead of crashing. 133 140 func (f *s3ReadFile) ReadAt(p []byte, off int64) (n int, err error) { 141 + if f.closed || f.reader == nil { 142 + return 0, os.ErrClosed 143 + } 134 144 return f.reader.ReadAt(p, off) 135 145 } 136 146 137 147 // Seek implements io.Seeker for billy.File 138 148 func (f *s3ReadFile) Seek(offset int64, whence int) (int64, error) { 149 + if f.closed || f.reader == nil { 150 + return 0, os.ErrClosed 151 + } 139 152 return f.reader.Seek(offset, whence) 140 153 } 141 154

+34

internal/s3fs/file_test.go

··· 3 3 import ( 4 4 "bytes" 5 5 "errors" 6 + "os" 6 7 "testing" 7 8 ) 9 + 10 + // TestS3ReadFileClosed locks in the billy contract go-git's 11 + // packfile.FSObject.Reader depends on: once a read file is closed, Read/ReadAt/Seek 12 + // must return an os.ErrClosed-matching error rather than dereferencing the nil 13 + // reader and panicking. FSObject probes a packed object's handle with a 1-byte 14 + // ReadAt and reopens the pack when it sees os.ErrClosed; the previous nil-deref 15 + // panic crashed the server when a cache-resident object outlived its pack handle 16 + // (e.g. a post-receive hook reading the just-pushed commit). 17 + func TestS3ReadFileClosed(t *testing.T) { 18 + newClosed := func() *s3ReadFile { 19 + f := &s3ReadFile{name: "k", reader: bytes.NewReader([]byte("hello"))} 20 + if err := f.Close(); err != nil { 21 + t.Fatalf("Close: %v", err) 22 + } 23 + return f 24 + } 25 + 26 + t.Run("ReadAt", func(t *testing.T) { 27 + if _, err := newClosed().ReadAt(make([]byte, 1), 0); !errors.Is(err, os.ErrClosed) { 28 + t.Fatalf("ReadAt after close: err = %v, want os.ErrClosed", err) 29 + } 30 + }) 31 + t.Run("Read", func(t *testing.T) { 32 + if _, err := newClosed().Read(make([]byte, 1)); !errors.Is(err, os.ErrClosed) { 33 + t.Fatalf("Read after close: err = %v, want os.ErrClosed", err) 34 + } 35 + }) 36 + t.Run("Seek", func(t *testing.T) { 37 + if _, err := newClosed().Seek(0, 0); !errors.Is(err, os.ErrClosed) { 38 + t.Fatalf("Seek after close: err = %v, want os.ErrClosed", err) 39 + } 40 + }) 41 + } 8 42 9 43 // TestS3WriteFileWrite locks in two invariants on the write path that a prior 10 44 // stub silently broke: every Write must append its bytes to the buffer, and

+430 -200

internal/s3fs/listingcache.go

··· 1 1 // listingcache.go caches directory listings so that Stat/Open of a path whose 2 2 // parent folder has been listed can be answered without an S3 round-trip: an 3 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. 4 + // free, and a present file is served from the head cache — seeded with each 5 + // file's size/mtime straight from the same ListObjectsV2 — so it too is free 6 + // unless the caller needs the x-amz-meta-* user metadata a listing can't carry. 6 7 // 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: 8 + // Everything is held in process-local sync.Maps with a conventional per-entry 9 + // TTL. There is no cross-process sharing: it is a plain in-memory cache. Two 10 + // mechanisms keep it correct: 11 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. 12 + // - per-entry expiry (expires = now + TTL): a stale entry is ignored once its 13 + // TTL passes, bounding how long a write this process can't see stays hidden. 14 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. 15 + // prefix (and its ancestors): an entry whose stored generation no longer 16 + // matches is ignored, so the next read re-lists and sees a local write 17 + // immediately (read-after-write correctness). 18 + // 19 + // Subtree caching (RecursivePrefixes). For namespaces that are bounded and 20 + // listed folder-by-folder by callers — refs/ above all — a delimited listing per 21 + // folder is wasteful: one delimiter-less ListObjectsV2 over refs/ returns the 22 + // whole subtree, from which every descendant folder's listing (and every 23 + // negative lookup beneath it) is synthesised in memory. So a prefix at or under a 24 + // recursive root is served from a single cached subtree scan instead of a listing 25 + // per folder. The trade-offs: a subtree write must invalidate every ancestor 26 + // (invalidate walks up), and a subtree larger than MaxSubtreeKeys is abandoned — 27 + // the cache records that and falls back to delimited per-folder listing. 18 28 package s3fs 19 29 20 30 import ( 21 31 "context" 22 - "encoding/json" 23 32 "errors" 24 33 "log/slog" 25 - "net/http" 26 - "strconv" 34 + "sort" 27 35 "strings" 28 36 "sync" 37 + "sync/atomic" 29 38 "time" 30 39 31 40 "github.com/aws/aws-sdk-go-v2/aws" 32 41 "github.com/aws/aws-sdk-go-v2/service/s3" 33 42 "github.com/aws/smithy-go" 34 - "github.com/golang/groupcache" 43 + "golang.org/x/sync/singleflight" 35 44 ) 36 45 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 46 // childKind distinguishes the two things a listing can report under a prefix. 43 47 type childKind uint8 44 48 ··· 47 51 kindDir 48 52 ) 49 53 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. 54 + // childEntry is one immediate child of a listed prefix. Name is the child's base 55 + // name (no separator). Size/Mtime are populated for files only; dirs (S3 common 56 + // prefixes) carry zero values. 53 57 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 + Name string 59 + Kind childKind 60 + Size int64 61 + Mtime int64 // unix nanoseconds 58 62 } 59 63 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). 64 + // headData is the size/mtime/metadata an s3.HeadObjectOutput is rebuilt from. 65 + // Size and Mtime can be seeded straight from a ListObjectsV2 entry; Meta (the 66 + // x-amz-meta-* user metadata for the Unix-metadata feature) only comes from an 67 + // actual HeadObject, so a listing-seeded entry leaves it nil — see hasMeta. 63 68 type headData struct { 64 - Size int64 `json:"s"` 65 - Mtime int64 `json:"m"` // unix nanoseconds 66 - Meta map[string]string `json:"u,omitempty"` 69 + Size int64 // bytes 70 + Mtime int64 // unix nanoseconds 71 + Meta map[string]string // x-amz-meta-* user metadata; nil when listing-seeded 67 72 } 68 73 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" 74 + // headCacheEntry is one object's cached head metadata. A read is a hit only when 75 + // the entry is unexpired and its gen still matches the parent prefix's current 76 + // generation — so a local write under the prefix (which bumps that generation) 77 + // invalidates every cached head beneath it without a map scan. hasMeta is false 78 + // for entries seeded from a listing (size/mtime only); a caller that needs the 79 + // user metadata treats those as a miss and fills via a real HeadObject. 80 + type headCacheEntry struct { 81 + data headData 82 + gen uint64 83 + expires time.Time 84 + hasMeta bool 85 + } 86 + 87 + // listingEntry is one folder's cached children, tagged with the prefix's 88 + // generation at fill time and an expiry. Same hit rule as headCacheEntry. 89 + type listingEntry struct { 90 + entries []childEntry 91 + gen uint64 92 + expires time.Time 93 + } 94 + 95 + // subtreeObject is one object in a recursive subtree scan: its full canonical 96 + // key plus the size/mtime ListObjectsV2 returns, enough to synthesise any 97 + // descendant folder's listing and to seed the head cache. 98 + type subtreeObject struct { 99 + Key string 100 + Size int64 101 + Mtime int64 // unix nanoseconds 102 + } 103 + 104 + // subtreeData is a recursive listing of a root prefix. When Truncated it hit 105 + // MaxSubtreeKeys and is unsafe for negative lookups, so callers fall back to 106 + // delimited per-folder listing for that root. 107 + type subtreeData struct { 108 + Objects []subtreeObject 109 + Truncated bool 110 + } 111 + 112 + // subtreeEntry is a cached subtreeData, tagged with the root's generation and an 113 + // expiry. 114 + type subtreeEntry struct { 115 + data subtreeData 116 + gen uint64 117 + expires time.Time 118 + } 72 119 73 120 // CacheConfig configures a ListingCache. TTL must be > 0 (callers gate the whole 74 121 // feature on it). The remaining fields have sane zero-value behaviour. 75 122 type CacheConfig struct { 76 - TTL time.Duration // key window; bounds cross-process staleness 123 + TTL time.Duration // per-entry lifetime; bounds negative staleness 77 124 RefreshInterval time.Duration // warmer tick; <=0 disables the warmer 78 125 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 126 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. 127 + // DisableHeadPrefetch turns off seeding the head cache from listings (the 128 + // size/mtime of every file in a folder, taken straight from ListObjectsV2). 129 + // The zero value keeps seeding enabled. 85 130 DisableHeadPrefetch bool 86 131 87 - Self string // this node's groupcache base URL ("" → single-process) 88 - Peers []string // peer base URLs, including Self, when sharing 132 + // RecursivePrefixes are namespaces served from one delimiter-less subtree 133 + // scan rather than a delimited listing per folder (see the package comment). 134 + // Each is normalised to end in "/". A nil slice defaults to {"refs/"}; an 135 + // explicit empty non-nil slice disables subtree caching entirely. 136 + RecursivePrefixes []string 137 + // MaxSubtreeKeys caps a subtree scan: past it the subtree is abandoned and 138 + // that root falls back to delimited per-folder listing (<=0 → 50000). 139 + MaxSubtreeKeys int 89 140 } 90 141 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 142 + // ListingCache is a process-local cache of directory listings, recursive 143 + // subtrees, and per-object head metadata, with per-entry TTL and per-prefix 144 + // generation invalidation. It is safe for concurrent use and is shared by 93 145 // pointer across an S3FS and all of its Chroot children. 94 146 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 147 ttl time.Duration 99 148 cfg CacheConfig 149 + client s3Client 150 + bucket string 151 + separator string 152 + roots []string // normalised RecursivePrefixes, longest first 100 153 101 - clock func() time.Time // overridable in tests 102 - prefetchSem chan struct{} // bounds background head precaches 154 + clock func() time.Time // overridable in tests 155 + 156 + // The three caches. Each value is the *Entry type above; reads honour the 157 + // entry's generation and expiry. sf dedupes concurrent listing/subtree fills, 158 + // headSF the head fills, so a fan-out of identical lookups lists S3 once. 159 + listings sync.Map // prefix → listingEntry 160 + subtrees sync.Map // root → subtreeEntry 161 + heads sync.Map // object key → headCacheEntry 162 + sf singleflight.Group 163 + headSF singleflight.Group 164 + 165 + hits, misses atomic.Int64 // listing + subtree cache outcomes, for metrics 103 166 104 167 mu sync.Mutex 105 168 gens map[string]uint64 // per-prefix local generation 106 169 seen map[string]time.Time // prefixes accessed → driven by the warmer 107 170 } 108 171 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. 172 + // NewListingCache builds a process-local cache. client/bucket/separator are the 173 + // root filesystem's, so cached prefixes are full-canonical keys that every 174 + // Chroot view agrees on. 113 175 func NewListingCache(cfg CacheConfig, client s3Client, bucket, separator string) *ListingCache { 114 - if cfg.Name == "" { 115 - cfg.Name = DefaultGroupName 176 + if cfg.MaxSubtreeKeys <= 0 { 177 + cfg.MaxSubtreeKeys = 50000 116 178 } 117 - if cfg.SizeBytes <= 0 { 118 - cfg.SizeBytes = 64 << 20 179 + if cfg.RecursivePrefixes == nil { 180 + cfg.RecursivePrefixes = []string{"refs/"} 119 181 } 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{}, 182 + return &ListingCache{ 183 + ttl: cfg.TTL, 184 + cfg: cfg, 185 + client: client, 186 + bucket: bucket, 187 + separator: separator, 188 + roots: normalizeRoots(cfg.RecursivePrefixes), 189 + clock: time.Now, 190 + gens: map[string]uint64{}, 191 + seen: map[string]time.Time{}, 127 192 } 193 + } 128 194 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 195 + // list returns the immediate children of prefix. prefix is a full-canonical key 196 + // prefix: "" for the bucket root, otherwise ending in the separator. A prefix at 197 + // or under a recursive root is synthesised from one cached subtree scan; 198 + // otherwise it is a delimited per-folder listing. 199 + func (c *ListingCache) list(ctx context.Context, prefix string) ([]childEntry, error) { 200 + c.touch(prefix) 201 + 202 + if root, ok := c.recursiveRoot(prefix); ok { 203 + st, err := c.subtree(ctx, root) 204 + // A complete subtree answers the folder in memory. On error or a 205 + // truncated (oversized) subtree, fall through to delimited listing. 206 + if err == nil && !st.Truncated { 207 + return synthesizeListing(st.Objects, prefix), nil 134 208 } 135 - data, err := json.Marshal(entries) 136 - if err != nil { 137 - return err 209 + } 210 + 211 + return c.listFolder(ctx, prefix) 212 + } 213 + 214 + // listFolder serves a single delimited folder listing from the cache, filling it 215 + // from S3 on a miss (deduped via singleflight). 216 + func (c *ListingCache) listFolder(ctx context.Context, prefix string) ([]childEntry, error) { 217 + gen := c.gen(prefix) 218 + if v, ok := c.listings.Load(prefix); ok { 219 + e := v.(listingEntry) 220 + if e.gen == gen && c.clock().Before(e.expires) { 221 + c.hits.Add(1) 222 + return e.entries, nil 138 223 } 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) 224 + } 225 + c.misses.Add(1) 226 + return c.fillFolder(ctx, prefix, gen) 227 + } 146 228 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) 229 + // fillFolder lists prefix from S3, stores the result, and seeds the head cache. 230 + // Concurrent fills for the same prefix are coalesced; the warmer also calls it 231 + // to force a refresh. 232 + func (c *ListingCache) fillFolder(ctx context.Context, prefix string, gen uint64) ([]childEntry, error) { 233 + v, err, _ := c.sf.Do("L\x00"+prefix, func() (any, error) { 234 + entries, err := listChildren(ctx, c.client, c.bucket, c.separator, prefix) 152 235 if err != nil { 153 - return err 236 + return nil, err 154 237 } 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) 238 + c.listings.Store(prefix, listingEntry{entries: entries, gen: gen, expires: c.clock().Add(c.ttl)}) 239 + // ListObjectsV2 already carries every file's size and mtime, so warm the 240 + // head cache from it without a single extra HeadObject. 241 + c.seedHeads(prefix, entries) 242 + return entries, nil 164 243 }) 165 - c.headGroup = groupcache.NewGroup(cfg.Name+"-heads", cfg.SizeBytes, headGetter) 244 + if err != nil { 245 + return nil, err 246 + } 247 + return v.([]childEntry), nil 248 + } 166 249 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} 250 + // subtree returns the recursive listing of root from the cache, filling it via 251 + // one delimiter-less ListObjectsV2 scan on a miss (deduped via singleflight). 252 + func (c *ListingCache) subtree(ctx context.Context, root string) (subtreeData, error) { 253 + gen := c.gen(root) 254 + if v, ok := c.subtrees.Load(root); ok { 255 + e := v.(subtreeEntry) 256 + if e.gen == gen && c.clock().Before(e.expires) { 257 + c.hits.Add(1) 258 + return e.data, nil 172 259 } 173 - c.pool.Set(peers...) 174 260 } 175 - 176 - return c 261 + c.misses.Add(1) 262 + return c.fillSubtree(ctx, root, gen) 177 263 } 178 264 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 265 + // fillSubtree scans root recursively, stores the result, and (when complete) 266 + // seeds the head cache for every file under it. 267 + func (c *ListingCache) fillSubtree(ctx context.Context, root string, gen uint64) (subtreeData, error) { 268 + v, err, _ := c.sf.Do("S\x00"+root, func() (any, error) { 269 + objs, truncated, err := listSubtree(ctx, c.client, c.bucket, root, c.cfg.MaxSubtreeKeys) 270 + if err != nil { 271 + return subtreeData{}, err 272 + } 273 + data := subtreeData{Objects: objs, Truncated: truncated} 274 + c.subtrees.Store(root, subtreeEntry{data: data, gen: gen, expires: c.clock().Add(c.ttl)}) 275 + // A complete subtree seeds the head cache for every file under the root 276 + // in one shot; a truncated one is untrustworthy, so leave heads to the 277 + // per-folder fallback path. 278 + if !truncated { 279 + c.seedSubtreeHeads(objs) 280 + } 281 + return data, nil 282 + }) 283 + if err != nil { 284 + return subtreeData{}, err 192 285 } 193 - return entries, nil 286 + return v.(subtreeData), nil 194 287 } 195 288 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) 289 + // recursiveRoot reports the longest configured recursive root at or above prefix. 290 + func (c *ListingCache) recursiveRoot(prefix string) (string, bool) { 291 + for _, r := range c.roots { // longest first 292 + if prefix == r || strings.HasPrefix(prefix, r) { 293 + return r, true 294 + } 295 + } 296 + return "", false 201 297 } 202 298 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) 299 + // synthesizeListing builds the immediate children of prefix from a subtree's 300 + // flat object set: a key whose remainder after prefix contains a separator is a 301 + // child directory (deduped), otherwise a child file. Order follows the objects' 302 + // (lexicographic) S3 order, dirs reported before files like listChildren. 303 + func synthesizeListing(objs []subtreeObject, prefix string) []childEntry { 304 + seenDir := map[string]bool{} 305 + var dirs, files []childEntry 306 + for _, o := range objs { 307 + if !strings.HasPrefix(o.Key, prefix) { 308 + continue 309 + } 310 + rest := o.Key[len(prefix):] 311 + if rest == "" { 312 + continue // the prefix's own placeholder object 313 + } 314 + if i := strings.IndexByte(rest, '/'); i >= 0 { 315 + name := rest[:i] 316 + if name == "" || seenDir[name] { 317 + continue 318 + } 319 + seenDir[name] = true 320 + dirs = append(dirs, childEntry{Name: name, Kind: kindDir}) 321 + continue 322 + } 323 + files = append(files, childEntry{Name: rest, Kind: kindFile, Size: o.Size, Mtime: o.Mtime}) 324 + } 325 + return append(dirs, files...) 210 326 } 211 327 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] 328 + // normalizeRoots cleans recursive-prefix config: drops blanks, ensures a 329 + // trailing separator, dedupes, and sorts longest-first so recursiveRoot returns 330 + // the most specific match. 331 + func normalizeRoots(prefixes []string) []string { 332 + seen := map[string]bool{} 333 + var out []string 334 + for _, p := range prefixes { 335 + if p == "" { 336 + continue 337 + } 338 + if !strings.HasSuffix(p, "/") { 339 + p += "/" 340 + } 341 + if seen[p] { 342 + continue 343 + } 344 + seen[p] = true 345 + out = append(out, p) 217 346 } 218 - return key 347 + sort.Slice(out, func(i, j int) bool { return len(out[i]) > len(out[j]) }) 348 + return out 219 349 } 220 350 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 { 351 + // headInfo returns the object's head metadata, served from the head cache 352 + // (seeded from listings, or filled on demand here). needMeta reports whether the 353 + // caller requires the x-amz-meta-* user metadata: when true a listing-seeded 354 + // entry (size/mtime only) is insufficient and a real HeadObject is issued. 355 + func (c *ListingCache) headInfo(ctx context.Context, objKey string, needMeta bool) (*s3.HeadObjectOutput, error) { 356 + d, err := c.headLoad(ctx, objKey, needMeta) 357 + if err != nil { 230 358 return nil, err 231 359 } 232 360 return &s3.HeadObjectOutput{ ··· 236 364 }, nil 237 365 } 238 366 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) { 367 + // headLoad returns objKey's head metadata from the head cache. An entry hits only 368 + // while unexpired, still tagged with its parent prefix's current generation (so 369 + // invalidate's bump drops every cached head under the prefix), and — when 370 + // needMeta — carrying user metadata. A miss fills via a single deduped HeadObject. 371 + func (c *ListingCache) headLoad(ctx context.Context, objKey string, needMeta bool) (headData, error) { 372 + prefix, _ := splitKey(objKey) 373 + gen := c.gen(prefix) 374 + if v, ok := c.heads.Load(objKey); ok { 375 + e := v.(headCacheEntry) 376 + if e.gen == gen && c.clock().Before(e.expires) && (e.hasMeta || !needMeta) { 377 + return e.data, nil 378 + } 379 + } 380 + 381 + // Miss: dedupe concurrent fills for the same key so only one HeadObject goes 382 + // to S3. 383 + v, err, _ := c.headSF.Do(objKey, func() (any, error) { 384 + start := time.Now() 385 + ho, err := c.client.HeadObject(ctx, &s3.HeadObjectInput{Bucket: &c.bucket, Key: &objKey}) 386 + observeS3("HeadObject", start, err) 387 + if err != nil { 388 + return headData{}, err 389 + } 390 + d := headData{ 391 + Size: aws.ToInt64(ho.ContentLength), 392 + Mtime: aws.ToTime(ho.LastModified).UnixNano(), 393 + Meta: ho.Metadata, 394 + } 395 + c.heads.Store(objKey, headCacheEntry{data: d, gen: gen, expires: c.clock().Add(c.ttl), hasMeta: true}) 396 + return d, nil 397 + }) 398 + if err != nil { 399 + return headData{}, err 400 + } 401 + return v.(headData), nil 402 + } 403 + 404 + // seedHeads warms the head cache for every file in a freshly-listed folder 405 + // directly from the listing's size/mtime — no HeadObject round-trips. The 406 + // entries are marked metadata-less (hasMeta=false), so a later read that needs 407 + // user metadata still triggers a real HeadObject; a read that doesn't (the 408 + // common case, and objgitd's only case) is served entirely from this seed. 409 + func (c *ListingCache) seedHeads(prefix string, entries []childEntry) { 244 410 if c.cfg.DisableHeadPrefetch { 245 411 return 246 412 } 413 + gen := c.gen(prefix) 414 + exp := c.clock().Add(c.ttl) 247 415 for _, e := range entries { 248 416 if e.Kind != kindFile { 249 417 continue 250 418 } 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 419 + c.heads.Store(prefix+e.Name, headCacheEntry{ 420 + data: headData{Size: e.Size, Mtime: e.Mtime}, 421 + gen: gen, 422 + expires: exp, 423 + }) 424 + } 425 + } 426 + 427 + // seedSubtreeHeads seeds the head cache for every file in a complete subtree 428 + // scan, the recursive analogue of seedHeads. Each entry is tagged with its own 429 + // parent prefix's generation so per-folder invalidation still applies. 430 + func (c *ListingCache) seedSubtreeHeads(objs []subtreeObject) { 431 + if c.cfg.DisableHeadPrefetch { 432 + return 433 + } 434 + exp := c.clock().Add(c.ttl) 435 + for _, o := range objs { 436 + prefix, base := splitKey(o.Key) 437 + if base == "" { 438 + continue // directory placeholder 257 439 } 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) 440 + c.heads.Store(o.Key, headCacheEntry{ 441 + data: headData{Size: o.Size, Mtime: o.Mtime}, 442 + gen: c.gen(prefix), 443 + expires: exp, 444 + }) 265 445 } 266 446 } 267 447 ··· 283 463 return c.gens[prefix] 284 464 } 285 465 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. 466 + // invalidate bumps the local generation of prefix and every ancestor up to the 467 + // bucket root, so the next read re-lists. Entries under the old generation are 468 + // simply ignored (and swept at expiry). The ancestor walk is what lets a subtree 469 + // entry (cached under a coarse root) notice a write to any descendant folder — at 470 + // the cost of a broader blast radius: a write also re-lists the coarser folders 471 + // above it on their next read. 288 472 func (c *ListingCache) invalidate(prefix string) { 289 473 if c == nil { 290 474 return 291 475 } 476 + now := c.clock() 292 477 c.mu.Lock() 293 - c.gens[prefix]++ 294 - c.seen[prefix] = c.clock() 478 + for _, p := range ancestorPrefixes(prefix) { 479 + c.gens[p]++ 480 + c.seen[p] = now 481 + } 295 482 c.mu.Unlock() 296 483 } 297 484 485 + // ancestorPrefixes returns prefix and every parent prefix up to and including 486 + // the bucket root (""). Each non-root element ends in "/". 487 + func ancestorPrefixes(prefix string) []string { 488 + out := []string{prefix} 489 + for prefix != "" { 490 + p := strings.TrimSuffix(prefix, "/") 491 + if i := strings.LastIndex(p, "/"); i >= 0 { 492 + prefix = p[:i+1] 493 + } else { 494 + prefix = "" 495 + } 496 + out = append(out, prefix) 497 + } 498 + return out 499 + } 500 + 298 501 func (c *ListingCache) touch(prefix string) { 299 502 c.mu.Lock() 300 503 c.seen[prefix] = c.clock() 301 504 c.mu.Unlock() 302 505 } 303 506 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. 507 + // RunWarmer refreshes every recently-accessed prefix on each tick — re-listing 508 + // it so a hot entry never lapses into a client-visible miss — then sweeps expired 509 + // entries from all three caches (there is no LRU, so the warmer is what bounds 510 + // their growth). It evicts prefixes unused for longer than IdleTTL. It returns 511 + // when ctx is cancelled; it is a no-op when the refresh interval is non-positive. 308 512 func (c *ListingCache) RunWarmer(ctx context.Context) { 309 513 if c == nil || c.cfg.RefreshInterval <= 0 { 310 514 <-ctx.Done() ··· 335 539 } 336 540 c.mu.Unlock() 337 541 542 + // Refresh via the same routing as the read path: prefixes under a recursive 543 + // root refresh that root's subtree (deduped across siblings), the rest their 544 + // delimited folder listing. 545 + roots := map[string]bool{} 338 546 for _, p := range prefixes { 339 - var data []byte 340 - if err := c.group.Get(ctx, c.groupKey(p), groupcache.AllocatingByteSliceSink(&data)); err != nil { 547 + if root, ok := c.recursiveRoot(p); ok { 548 + roots[root] = true 549 + continue 550 + } 551 + if _, err := c.fillFolder(ctx, p, c.gen(p)); err != nil { 341 552 slog.Debug("listing-cache warm failed", "prefix", p, "err", err) 342 553 } 343 554 } 555 + for root := range roots { 556 + if _, err := c.fillSubtree(ctx, root, c.gen(root)); err != nil { 557 + slog.Debug("subtree-cache warm failed", "root", root, "err", err) 558 + } 559 + } 560 + 561 + c.sweepExpired(now) 344 562 } 345 563 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 564 + // sweepExpired drops expired entries from all three caches. Stale-generation 565 + // entries that have not yet expired are left; they are never read (generation 566 + // mismatch) and age out here once their TTL passes. 567 + func (c *ListingCache) sweepExpired(now time.Time) { 568 + c.listings.Range(func(k, v any) bool { 569 + if !now.Before(v.(listingEntry).expires) { 570 + c.listings.Delete(k) 571 + } 572 + return true 573 + }) 574 + c.subtrees.Range(func(k, v any) bool { 575 + if !now.Before(v.(subtreeEntry).expires) { 576 + c.subtrees.Delete(k) 577 + } 578 + return true 579 + }) 580 + c.heads.Range(func(k, v any) bool { 581 + if !now.Before(v.(headCacheEntry).expires) { 582 + c.heads.Delete(k) 583 + } 584 + return true 585 + }) 353 586 } 354 587 355 - // CacheStats is a flat snapshot of groupcache counters for export to metrics. 588 + // CacheStats is a snapshot of the cache's counters for export to metrics. 356 589 type CacheStats struct { 357 - Gets, CacheHits, Loads, LocalLoads, PeerLoads, LocalLoadErrs int64 358 - MainBytes, MainItems, MainEvictions int64 359 - HotBytes, HotItems int64 590 + Hits, Misses int64 591 + ListingItems, SubtreeItems, HeadItems int64 360 592 } 361 593 362 - // Stats snapshots the underlying groupcache counters. 594 + // Stats snapshots hit/miss counters and resident item counts. 363 595 func (c *ListingCache) Stats() CacheStats { 364 - if c == nil || c.group == nil { 596 + if c == nil { 365 597 return CacheStats{} 366 598 } 367 - main := c.group.CacheStats(groupcache.MainCache) 368 - hot := c.group.CacheStats(groupcache.HotCache) 369 599 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, 600 + Hits: c.hits.Load(), 601 + Misses: c.misses.Load(), 602 + ListingItems: syncMapLen(&c.listings), 603 + SubtreeItems: syncMapLen(&c.subtrees), 604 + HeadItems: syncMapLen(&c.heads), 381 605 } 606 + } 607 + 608 + func syncMapLen(m *sync.Map) int64 { 609 + var n int64 610 + m.Range(func(_, _ any) bool { n++; return true }) 611 + return n 382 612 } 383 613 384 614 // splitKey splits a canonical key into its parent prefix and base name. The

+132 -28

internal/s3fs/listingcache_test.go

··· 4 4 "bytes" 5 5 "context" 6 6 "errors" 7 - "fmt" 8 7 "io" 9 8 "io/fs" 10 9 "sort" ··· 132 131 return &s3.CompleteMultipartUploadOutput{}, nil 133 132 } 134 133 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. 134 + // newTestCache disables listing-driven head seeding so S3-call counts are 135 + // deterministic; TestListingCacheHeadSeed exercises it explicitly. 141 136 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", "/") 137 + return NewListingCache(CacheConfig{TTL: ttl, DisableHeadPrefetch: true}, stub, "bucket", "/") 144 138 } 145 139 146 140 func newCachedFS(t *testing.T, stub *stubClient, cache *ListingCache) billy.Filesystem { ··· 320 314 } 321 315 } 322 316 323 - func TestListingCacheHeadPrefetch(t *testing.T) { 317 + func TestListingCacheHeadSeed(t *testing.T) { 324 318 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", "/") 319 + cache := NewListingCache(CacheConfig{TTL: time.Hour}, stub, "bucket", "/") 327 320 fsys := newCachedFS(t, stub, cache) 328 321 329 - // A single listing fill (here triggered by an absent lookup) prefetches the 330 - // HeadObject for every file in the folder, in the background. 322 + // A single listing fill (here triggered by an absent lookup) seeds the head 323 + // cache for every file in the folder straight from the ListObjectsV2 data — 324 + // no HeadObject round-trips at all. 331 325 if _, err := fsys.Stat("objects/ab/nope"); !errors.Is(err, fs.ErrNotExist) { 332 326 t.Fatalf("Stat: %v", err) 333 327 } 328 + if got := stub.heads.Load(); got != 0 { 329 + t.Fatalf("seeding issued HeadObjects: got %d, want 0", got) 330 + } 334 331 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()) 332 + // Stat of a seeded file is served from the head cache: still no HeadObject, 333 + // and it reports a file with the listed size. 334 + fi, err := fsys.Stat("objects/ab/f1") 335 + if err != nil { 336 + t.Fatalf("Stat f1: %v", err) 337 + } 338 + if fi.IsDir() { 339 + t.Fatalf("Stat f1: reported a directory") 340 + } 341 + if got := stub.heads.Load(); got != 0 { 342 + t.Fatalf("seeded Stat did a HeadObject: got %d, want 0", got) 343 + } 344 + } 345 + 346 + func TestListingCacheSubtreeCollapsesFolders(t *testing.T) { 347 + stub := newStub("refs/heads/main", "refs/heads/dev", "refs/tags/v1") 348 + cache := newTestCache(stub, time.Hour) // default RecursivePrefixes = {"refs/"} 349 + fsys := newCachedFS(t, stub, cache) 350 + 351 + // The first touch of any refs/ folder scans the whole refs/ subtree once. 352 + entries, err := fsys.ReadDir("refs/heads") 353 + if err != nil { 354 + t.Fatalf("ReadDir refs/heads: %v", err) 355 + } 356 + if len(entries) != 2 { 357 + t.Fatalf("refs/heads entries = %d, want 2", len(entries)) 358 + } 359 + if stub.lists.Load() != 1 { 360 + t.Fatalf("first refs read lists = %d, want 1", stub.lists.Load()) 361 + } 362 + 363 + // A different refs/ folder is served from that same subtree: no new S3. 364 + l0 := stub.lists.Load() 365 + tags, err := fsys.ReadDir("refs/tags") 366 + if err != nil { 367 + t.Fatalf("ReadDir refs/tags: %v", err) 368 + } 369 + if len(tags) != 1 { 370 + t.Fatalf("refs/tags entries = %d, want 1", len(tags)) 371 + } 372 + if stub.lists.Load() != l0 { 373 + t.Fatalf("second refs folder re-listed: %d->%d", l0, stub.lists.Load()) 374 + } 375 + 376 + // refs/ itself synthesises its child directories from the subtree. 377 + top, err := fsys.ReadDir("refs") 378 + if err != nil { 379 + t.Fatalf("ReadDir refs: %v", err) 380 + } 381 + if len(top) != 2 { 382 + t.Fatalf("refs entries = %d, want 2 (heads, tags)", len(top)) 383 + } 384 + for _, e := range top { 385 + if !e.IsDir() { 386 + t.Fatalf("refs child %q not a directory", e.Name()) 340 387 } 341 - time.Sleep(time.Millisecond) 342 388 } 343 - if got := stub.heads.Load(); got != 2 { 344 - t.Fatalf("prefetch heads = %d, want 2", got) 389 + 390 + // And a negative lookup anywhere under refs/ is free. 391 + if _, err := fsys.Stat("refs/heads/missing"); !errors.Is(err, fs.ErrNotExist) { 392 + t.Fatalf("Stat missing: want ErrNotExist, got %v", err) 393 + } 394 + if stub.lists.Load() != l0 { 395 + t.Fatalf("negative lookup did S3: %d->%d", l0, stub.lists.Load()) 345 396 } 397 + } 346 398 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) 399 + func TestListingCacheSubtreeInvalidate(t *testing.T) { 400 + stub := newStub("refs/heads/main") 401 + cache := newTestCache(stub, time.Hour) 402 + fsys := newCachedFS(t, stub, cache) 403 + 404 + // Warm the whole refs/ subtree by reading one folder. 405 + if _, err := fsys.ReadDir("refs/heads"); err != nil { 406 + t.Fatalf("ReadDir refs/heads: %v", err) 351 407 } 352 - if stub.heads.Load() != h0 { 353 - t.Fatalf("prewarmed Stat did a HeadObject: %d->%d", h0, stub.heads.Load()) 408 + if stub.lists.Load() != 1 { 409 + t.Fatalf("warm lists = %d, want 1", stub.lists.Load()) 410 + } 411 + 412 + // Write a ref into a *sibling* folder. Ancestor invalidation must move the 413 + // refs/ subtree key so the next read re-scans and sees it. 414 + f, err := fsys.Create("refs/tags/v1") 415 + if err != nil { 416 + t.Fatalf("Create: %v", err) 417 + } 418 + if _, err := f.Write([]byte("deadbeef")); err != nil { 419 + t.Fatalf("Write: %v", err) 420 + } 421 + if err := f.Close(); err != nil { 422 + t.Fatalf("Close: %v", err) 423 + } 424 + 425 + tags, err := fsys.ReadDir("refs/tags") 426 + if err != nil { 427 + t.Fatalf("ReadDir refs/tags after write: %v", err) 428 + } 429 + if len(tags) != 1 { 430 + t.Fatalf("refs/tags entries = %d, want 1", len(tags)) 431 + } 432 + if stub.lists.Load() < 2 { 433 + t.Fatalf("subtree not re-scanned after sibling write: lists = %d", stub.lists.Load()) 434 + } 435 + } 436 + 437 + func TestListingCacheSubtreeTruncationFallback(t *testing.T) { 438 + stub := newStub("refs/heads/a", "refs/heads/b", "refs/heads/c") 439 + cache := NewListingCache(CacheConfig{ 440 + TTL: time.Hour, 441 + DisableHeadPrefetch: true, 442 + MaxSubtreeKeys: 2, // 3 refs exceed the cap → subtree abandoned 443 + }, stub, "bucket", "/") 444 + fsys := newCachedFS(t, stub, cache) 445 + 446 + // The oversized subtree scan is abandoned; the folder is still served 447 + // correctly, by falling back to a delimited per-folder listing. 448 + entries, err := fsys.ReadDir("refs/heads") 449 + if err != nil { 450 + t.Fatalf("ReadDir refs/heads: %v", err) 451 + } 452 + if len(entries) != 3 { 453 + t.Fatalf("entries = %d, want 3", len(entries)) 454 + } 455 + // One truncated subtree scan + one delimited fallback listing. 456 + if stub.lists.Load() != 2 { 457 + t.Fatalf("fallback lists = %d, want 2", stub.lists.Load()) 354 458 } 355 459 } 356 460

Configure Feed

Configure Feed