bootstrap: stream commit-graph fetch to bound planning memory

f5cc40c→main·

nodo·1mo ago·7 files·+892 added/-62 removed

Replace the per-branch memory.NewStorage() planning fetch with a streaming extractor that pulls only (commit -> parent hashes) tuples from the tree:0-filtered pack.

Previously, bootstrap planning materialized the entire commit set (~1.4M decoded commits for linux, ~5 GiB) just to read each commit's ParentHashes. The new path reads the pack incrementally, extracting parents from each commit as it arrives and discarding the bytes; a bounded LRU keeps recent objects in memory only long enough to resolve deltas. Planning then walks the resulting parents map directly.

Measured against replicate --all-refs https://github.com/torvalds/linux.git into a local git-http-backend target:

Changes

7

• internal

• gitproto

• Acommitparents.go+300

• Acommitparents_test.go+197

• planner

• Mcheckpoint.go+124

• Acheckpoint_from_parents_test.go+131

• strategy/bootstrap

• Mbootstrap.go+20/-44

• Mbootstrap_test.go+42/-14

package gitproto

import (
    "bytes"
    "container/list"
    "errors"
    "fmt"
    "io"
    "os"

"github.com/go-git/go-git/v6/plumbing"
    "github.com/go-git/go-git/v6/plumbing/format/packfile"
    "github.com/go-git/go-git/v6/plumbing/storer"
)

// DefaultCommitParentsCacheBytes is the default sliding-cache size for
// ExtractCommitParents. Sized to comfortably cover delta locality for
// tree:0-filtered packs generated with git's default window/depth
// (typically tens of objects of working set).
const DefaultCommitParentsCacheBytes = 64 * 1024 * 1024

// ExtractCommitParents reads a Git packfile from r and returns a map
// of commit hash → parent hashes for every commit object encountered.
// Non-commit objects (tags, etc.) and the raw bytes of commits are
// discarded after parent extraction, so peak memory is dominated by
// the result map plus a small bounded delta-resolution cache.
//
// Intended for tree:0-filtered packs where the only object kinds are
// commits and (occasionally) annotated tags. Input is the raw packfile
// stream (e.g. demuxed sideband from a v2 fetch response).
func ExtractCommitParents(r io.Reader) (map[plumbing.Hash][]plumbing.Hash, error) {
    return ExtractCommitParentsWithCache(r, DefaultCommitParentsCacheBytes)
}

// ExtractCommitParentsWithCache is like ExtractCommitParents but lets
// the caller cap the delta-resolution cache size. On cache miss the
// underlying pack parser returns ErrObjectNotFound; raise the limit
// and retry, or fall back to the in-memory store path.
func ExtractCommitParentsWithCache(r io.Reader, maxCacheBytes int) (map[plumbing.Hash][]plumbing.Hash, error) {
    if maxCacheBytes <= 0 {
        maxCacheBytes = DefaultCommitParentsCacheBytes
    }

// go-git's pack parser forces lowMemoryMode = false when the input
    // is not seekable (parser.go:83-85), which causes the scanner to
    // buffer every inflated object in memory until Parse returns
    // (scanner.go:427) — undoing the savings the custom storer aims
    // for. We spill the pack to a temp file so the parser sees a
    // seekable source, enables low-memory mode, and releases object
    // content as it flows through the storer. Already-seekable inputs
    // (e.g. tests using bytes.Reader) skip the spill.
    seekable, ok := r.(io.ReadSeeker)
    if !ok {
        tmp, err := os.CreateTemp("", "gitsync-cgparse-*.pack")
        if err != nil {
            return nil, fmt.Errorf("create temp pack: %w", err)
        }
        defer os.Remove(tmp.Name())
        defer tmp.Close()

if _, err := io.Copy(tmp, r); err != nil {
            return nil, fmt.Errorf("spill pack to temp: %w", err)
        }
        if _, err := tmp.Seek(0, io.SeekStart); err != nil {
            return nil, fmt.Errorf("seek temp pack: %w", err)
        }
        seekable = tmp
    }

store := newCommitParentsStorer(maxCacheBytes)
    p := packfile.NewParser(seekable, packfile.WithStorage(store))
    if _, err := p.Parse(); err != nil {
        return nil, fmt.Errorf("parse packfile: %w", err)
    }
    return store.parents, nil
}

// commitParentsStorer is a minimal storer.EncodedObjectStorer that
// captures commit parent metadata as objects flow through the parser,
// and serves recent objects from a bounded LRU cache when the parser
// needs to resolve a delta base.
//
// It implements packfile.LowMemoryCapable to tell the parser it can
// release its inflated-content buffers as soon as objects are handed
// off here — we only retain a small working set in the cache.
type commitParentsStorer struct {
    parents map[plumbing.Hash][]plumbing.Hash
    cache   *deltaCache
    hasher  *plumbing.ObjectHasher
}

func newCommitParentsStorer(maxBytes int) *commitParentsStorer {
    return &commitParentsStorer{
        parents: make(map[plumbing.Hash][]plumbing.Hash),
        cache:   newDeltaCache(maxBytes),
        hasher:  plumbing.FromObjectFormat(""),
    }
}

// LowMemoryMode tells the parser we don't need it to keep object
// content in its internal cache; we handle delta-base retrieval via
// EncodedObject below.
func (s *commitParentsStorer) LowMemoryMode() bool { return true }

// RawObjectWriter returns a sink for one inflated object. The writer's
// Close captures commit parents and caches the bytes for potential
// future delta lookups.
func (s *commitParentsStorer) RawObjectWriter(typ plumbing.ObjectType, sz int64) (io.WriteCloser, error) {
    return &commitParentsWriter{
        s:   s,
        typ: typ,
        buf: bytes.NewBuffer(make([]byte, 0, sz)),
    }, nil
}

type commitParentsWriter struct {
    s   *commitParentsStorer
    typ plumbing.ObjectType
    buf *bytes.Buffer
}

func (w *commitParentsWriter) Write(p []byte) (int, error) {
    n, err := w.buf.Write(p)
    if err != nil {
        return n, fmt.Errorf("buffer object bytes: %w", err)
    }
    return n, nil
}

func (w *commitParentsWriter) Close() error {
    content := w.buf.Bytes()
    hash, err := w.s.hasher.Compute(w.typ, content)
    if err != nil {
        return fmt.Errorf("hash %s object (%d bytes): %w", w.typ, len(content), err)
    }
    if w.typ == plumbing.CommitObject {
        w.s.parents[hash] = parseCommitParents(content)
    }
    // Detach the byte slice from the buffer so the buffer can be GC'd
    // independently. The cache may keep this entry until evicted.
    w.s.cache.put(hash, w.typ, content)
    return nil
}

// EncodedObject is queried by the parser when resolving deltas in
// low-memory mode. Serves from the bounded LRU cache. The interface
// contract calls for matching both hash *and* type, except when the
// caller passes AnyObject as a wildcard.
func (s *commitParentsStorer) EncodedObject(typ plumbing.ObjectType, hash plumbing.Hash) (plumbing.EncodedObject, error) {
    entry, ok := s.cache.get(hash)
    if !ok {
        return nil, plumbing.ErrObjectNotFound
    }
    if typ != plumbing.AnyObject && entry.typ != typ {
        return nil, plumbing.ErrObjectNotFound
    }
    obj := plumbing.NewMemoryObject(s.hasher)
    obj.SetType(entry.typ)
    obj.SetSize(int64(len(entry.content)))
    w, err := obj.Writer()
    if err != nil {
        return nil, fmt.Errorf("memory-object writer: %w", err)
    }
    if _, err := w.Write(entry.content); err != nil {
        _ = w.Close()
        return nil, fmt.Errorf("memory-object write: %w", err)
    }
    if err := w.Close(); err != nil {
        return nil, fmt.Errorf("memory-object close: %w", err)
    }
    return obj, nil
}

// --- Unused interface methods ------------------------------------------------
//
// EncodedObjectStorer requires more than the two methods the parser
// actually exercises in low-memory mode. We stub the rest with explicit
// errors so any future caller that depends on them fails loudly rather
// than getting a misleading empty result.

func (s *commitParentsStorer) NewEncodedObject() plumbing.EncodedObject {
    return plumbing.NewMemoryObject(s.hasher)
}

func (s *commitParentsStorer) SetEncodedObject(plumbing.EncodedObject) (plumbing.Hash, error) {
    return plumbing.ZeroHash, errors.New("commitParentsStorer: SetEncodedObject not supported")
}

func (s *commitParentsStorer) IterEncodedObjects(plumbing.ObjectType) (storer.EncodedObjectIter, error) {
    return nil, errors.New("commitParentsStorer: IterEncodedObjects not supported")
}

func (s *commitParentsStorer) HasEncodedObject(plumbing.Hash) error {
    return errors.New("commitParentsStorer: HasEncodedObject not supported")
}

func (s *commitParentsStorer) EncodedObjectSize(plumbing.Hash) (int64, error) {
    return 0, errors.New("commitParentsStorer: EncodedObjectSize not supported")
}

func (s *commitParentsStorer) AddAlternate(string) error {
    return errors.New("commitParentsStorer: AddAlternate not supported")
}

// --- Delta cache -------------------------------------------------------------
type deltaCacheEntry struct {
    hash    plumbing.Hash
    typ     plumbing.ObjectType
    content []byte
    elem    *list.Element
}
type deltaCache struct {
    maxBytes int
    curBytes int
    byHash   map[plumbing.Hash]*deltaCacheEntry
    lru      *list.List // front = most recently used
}

func newDeltaCache(maxBytes int) *deltaCache {
    return &deltaCache{
        maxBytes: maxBytes,
        byHash:   make(map[plumbing.Hash]*deltaCacheEntry),
        lru:      list.New(),
    }
}

func (c *deltaCache) put(hash plumbing.Hash, typ plumbing.ObjectType, content []byte) {
    if existing, ok := c.byHash[hash]; ok {
        c.curBytes -= len(existing.content)
        c.lru.Remove(existing.elem)
        delete(c.byHash, hash)
    }
    entry := &deltaCacheEntry{hash: hash, typ: typ, content: content}
    entry.elem = c.lru.PushFront(entry)
    c.byHash[hash] = entry
    c.curBytes += len(content)
    for c.curBytes > c.maxBytes {
        back := c.lru.Back()
        if back == nil {
            return
        }
        oldest, ok := back.Value.(*deltaCacheEntry)
        if !ok {
            // container/list values are interface{}; only put() ever
            // inserts here and only with *deltaCacheEntry, so a
            // mismatch would indicate corruption. Bail rather than
            // loop forever (curBytes would never decrease).
            panic("deltaCache: corrupt LRU entry")
        }
        c.curBytes -= len(oldest.content)
        c.lru.Remove(back)
        delete(c.byHash, oldest.hash)
    }
}

func (c *deltaCache) get(hash plumbing.Hash) (*deltaCacheEntry, bool) {
    entry, ok := c.byHash[hash]
    if !ok {
        return nil, false
    }
    c.lru.MoveToFront(entry.elem)
    return entry, true
}

// --- Commit-header parsing ---------------------------------------------------
// parseCommitParents walks the header lines of a commit object and
// returns the hashes referenced by "parent <hash>\n" lines. Stops at
// the first empty line (end of headers) or at "author " (parents
// always appear before author in well-formed commits).
func parseCommitParents(content []byte) []plumbing.Hash {
    const prefix = "parent "
    var parents []plumbing.Hash
    rest := content
    for len(rest) > 0 {
        nl := bytes.IndexByte(rest, '\n')
        var line []byte
        if nl < 0 {
            line = rest
            rest = nil
        } else {
            line = rest[:nl]
            rest = rest[nl+1:]
        }
        if len(line) == 0 {
            return parents
        }
        switch {
        case bytes.HasPrefix(line, []byte(prefix)):
            if len(line) >= len(prefix)+40 {
                parents = append(parents, plumbing.NewHash(string(line[len(prefix):len(prefix)+40])))
            }
        case bytes.HasPrefix(line, []byte("author ")):
            return parents
        }
    }
    return parents
}