diff --git a/go.mod b/go.mod
index edf05f0..f1cb838 100644
--- a/go.mod
+++ b/go.mod
@@ -13,6 +13,7 @@ require (
github.com/yuin/goldmark v1.8.2
github.com/yuin/goldmark-highlighting/v2 v2.0.0-20230729083705-37449abec8cc
go.abhg.dev/goldmark/mermaid v0.6.0
+ golang.org/x/sync v0.20.0
gopkg.in/yaml.v3 v3.0.1
)
@@ -64,7 +65,6 @@ require (
github.com/wasilibs/go-re2 v1.3.0 // indirect
golang.org/x/exp v0.0.0-20251023183803-a4bb9ffd2546 // indirect
golang.org/x/net v0.55.0 // indirect
- golang.org/x/sync v0.20.0 // indirect
golang.org/x/sys v0.45.0 // indirect
golang.org/x/text v0.37.0 // indirect
google.golang.org/protobuf v1.36.10 // indirect
diff --git a/internal/repo/materialize.go b/internal/repo/materialize.go
index 9692dbb..e06ee03 100644
--- a/internal/repo/materialize.go
+++ b/internal/repo/materialize.go
@@ -16,6 +16,8 @@ import (
"path/filepath"
"strings"
+ "golang.org/x/sync/errgroup"
+
"github.com/atmin/gitmote/internal/meta"
"github.com/atmin/gitmote/internal/store"
)
@@ -133,16 +135,26 @@ func (mz *Materializer) repoDir(name string) (string, error) {
return filepath.Join(mz.root, filepath.FromSlash(name)), nil
}
+// hydrateConcurrency bounds the parallel object copies during hydration. Each
+// missing object is one S3 GET, and this is the cold-start / first-request cost
+// (docs/notes/object-hydration.md); fetching them concurrently turns N serial
+// round-trips into N/hydrateConcurrency. Kept modest to bound open sockets and
+// file descriptors against a large repo.
+const hydrateConcurrency = 16
+
// hydrateObjects copies every object under the repo's store prefix onto disk,
// mirroring the store's {repo}/objects/… layout into the bare repo's objects/
// dir. Objects are immutable and content-addressed, so an object already on
-// disk is left untouched.
+// disk is left untouched, and copies are independent — hydrateConcurrency run
+// in parallel. A copy error cancels the group and aborts the rest.
func (mz *Materializer) hydrateObjects(ctx context.Context, name, dir string) error {
prefix := name + "/objects/"
keys, err := mz.store.List(ctx, prefix)
if err != nil {
return err
}
+ g, ctx := errgroup.WithContext(ctx)
+ g.SetLimit(hydrateConcurrency)
for _, key := range keys {
rel := strings.TrimPrefix(key, name+"/") // e.g. "objects/ab/cdef…"
dst := filepath.Join(dir, filepath.FromSlash(rel))
@@ -151,11 +163,9 @@ func (mz *Materializer) hydrateObjects(ctx context.Context, name, dir string) er
} else if !errors.Is(err, os.ErrNotExist) {
return err
}
- if err := mz.copyObject(ctx, key, dst); err != nil {
- return err
- }
+ g.Go(func() error { return mz.copyObject(ctx, key, dst) })
}
- return nil
+ return g.Wait()
}
// copyObject streams one store object to dst, writing to a temp file first and
diff --git a/internal/repo/materialize_test.go b/internal/repo/materialize_test.go
index 0f59cf7..5e8d1c9 100644
--- a/internal/repo/materialize_test.go
+++ b/internal/repo/materialize_test.go
@@ -4,6 +4,7 @@ import (
"bytes"
"context"
"errors"
+ "fmt"
"io"
"os"
"os/exec"
@@ -260,3 +261,67 @@ func TestRepoDirRejectsUnsafeName(t *testing.T) {
}
}
}
+
+// TestMaterializeManyObjectsParallel drives hydrateObjects with far more loose
+// objects than hydrateConcurrency, so the bounded pool actually queues, and
+// asserts the concurrent copies still reproduce a byte-identical, fsck-clean
+// repo (no torn objects, no lost keys) at the right head.
+func TestMaterializeManyObjectsParallel(t *testing.T) {
+ ctx := context.Background()
+ m := openMeta(t)
+ s := store.NewMem()
+ const repoName = "many"
+
+ // One commit touching 60 distinct files → 60 blobs + trees + a commit, well
+ // past hydrateConcurrency (16), forcing the pool to block and drain.
+ src := t.TempDir()
+ git(t, src, "init", "-b", "main", ".")
+ for i := 0; i < 60; i++ {
+ name := fmt.Sprintf("f%02d.txt", i)
+ if err := os.WriteFile(filepath.Join(src, name), []byte(fmt.Sprintf("content %d\n", i)), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ }
+ git(t, src, "add", "-A")
+ git(t, src, "commit", "-m", "many files") // no repack: keep objects loose
+ branch := git(t, src, "rev-parse", "--abbrev-ref", "HEAD")
+ head := git(t, src, "rev-parse", "HEAD")
+
+ objRoot := filepath.Join(src, ".git", "objects")
+ loose := 0
+ err := filepath.WalkDir(objRoot, func(p string, d os.DirEntry, err error) error {
+ if err != nil || d.IsDir() {
+ return err
+ }
+ rel, err := filepath.Rel(objRoot, p)
+ if err != nil {
+ return err
+ }
+ data, err := os.ReadFile(p)
+ if err != nil {
+ return err
+ }
+ loose++
+ return s.Put(ctx, repoName+"/objects/"+filepath.ToSlash(rel), bytes.NewReader(data))
+ })
+ if err != nil {
+ t.Fatalf("seed objects: %v", err)
+ }
+ if loose <= hydrateConcurrency {
+ t.Fatalf("seeded %d objects, need > hydrateConcurrency (%d) to exercise queueing", loose, hydrateConcurrency)
+ }
+
+ r, err := m.CreateRepo(ctx, repoName, branch)
+ if err != nil {
+ t.Fatalf("CreateRepo: %v", err)
+ }
+ if err := m.CASRef(ctx, r.ID, "refs/heads/"+branch, meta.ZeroSHA, head); err != nil {
+ t.Fatalf("CASRef: %v", err)
+ }
+
+ dir, err := New(m, s, t.TempDir()).Materialize(ctx, repoName)
+ if err != nil {
+ t.Fatalf("Materialize: %v", err)
+ }
+ assertRepo(t, dir, branch, head)
+}