From b1fae8d5a2c1814d2e1aa6e50d01ccb7428f858c Mon Sep 17 00:00:00 2001 From: Xe Iaso Date: Fri, 25 Sep 2026 16:28:55 -0400 Subject: [PATCH 1/8] perf(storage/tigris): index pushed packs in bounded memory A push decoded its pack through a scratch go-git filesystem.Storage. That storage parses with no storage attached, so go-git's parser stays out of low-memory mode and keeps every inflated base and resolved delta until the parse ends. A push of golang/go (465 MiB pack) passed 21 GiB of heap and did not complete. Replace it with a small index-pack in indexpack.go. One pass stages, checksums, and indexes the pack as it arrives, with no bodies kept. A second pass resolves each delta tree depth first, so a body lives only while a delta below it needs it. Bodies over 8 MiB, or past a 64 MiB total, go to temp files. writePack hands the stream over through a new optional ReadPack method, which also removes a go-git Scanner that buffered every object body in full. golang/go now peaks at about 521 MiB of heap and takes 23 s. Xe/x goes from 516 MiB and 14.7 s to about 165 MiB and 1.3 s. Delta reuse is unchanged. This also fixes stored delta chains of exactly 50 links, which git makes by default and which reads refused. Reads now accept a chain of maxDeltaDepth links, and the writer stores an object whole rather than extend a stored chain past that, since clients can send 4095. Assisted-by: Claude Opus 5.5 via Claude Code Signed-off-by: Xe Iaso --- cmd/objgitd/receivepack.go | 17 + internal/storage/tigris/compress.go | 4 +- internal/storage/tigris/indexpack.go | 911 ++++++++++++++++++++++++++ internal/storage/tigris/packindex.go | 10 +- internal/storage/tigris/packwriter.go | 375 +++-------- 5 files changed, 1013 insertions(+), 304 deletions(-) create mode 100644 internal/storage/tigris/indexpack.go diff --git a/cmd/objgitd/receivepack.go b/cmd/objgitd/receivepack.go index 3982c4e..39af635 100644 --- a/cmd/objgitd/receivepack.go +++ b/cmd/objgitd/receivepack.go @@ -268,6 +268,17 @@ func writePack(st storage.Storer, rd io.Reader) error { return err } + // A writer that reads the pack itself finds its end in the same pass that + // stores it. The Scanner below cannot do that in bounded memory: over a + // reader that cannot seek, it buffers every object's whole inflated body. + if pr, ok := w.(packReader); ok { + if err := pr.ReadPack(rd); err != nil { + _ = w.Close() + return err + } + return w.Close() + } + sc := packfile.NewScanner(io.TeeReader(rd, w), sopts...) for sc.Scan() { } @@ -278,6 +289,12 @@ func writePack(st storage.Storer, rd io.Reader) error { return w.Close() } +// packReader is the optional PackfileWriter surface that reads exactly one +// packfile from r and stops at its trailer. The tigris storer has one. +type packReader interface { + ReadPack(r io.Reader) error +} + // sidebandProgress writes to the sideband progress channel (band 2), which the // git client renders as "remote: " lines. type sidebandProgress struct { diff --git a/internal/storage/tigris/compress.go b/internal/storage/tigris/compress.go index 77c41b7..96b1715 100644 --- a/internal/storage/tigris/compress.go +++ b/internal/storage/tigris/compress.go @@ -100,8 +100,8 @@ var ( // and every internal encoder state allocates its own match-history buffer (tens // of megabytes, never released), so the default puts a permanent floor under // retained heap that scales with core count and not with real demand. The value -// that matters is "expected simultaneous pushes", not "core count"; 4 matches -// deltaScanWorkers in spirit — a bound picked for resource reasons. Revisit it +// that matters is "expected simultaneous pushes", not "core count"; 4 is a +// bound picked for resource reasons. Revisit it // alongside the push cap that actually bounds demand. // // A package-level variable with a setter, not a Storer option: the encoder is a diff --git a/internal/storage/tigris/indexpack.go b/internal/storage/tigris/indexpack.go new file mode 100644 index 0000000..a4aef40 --- /dev/null +++ b/internal/storage/tigris/indexpack.go @@ -0,0 +1,911 @@ +package tigris + +import ( + "bufio" + "bytes" + "crypto/sha1" + "crypto/sha256" + "encoding/binary" + "errors" + "fmt" + "hash" + "io" + "os" + "sort" + "sync" + + "github.com/go-git/go-git/v6/plumbing" + formatcfg "github.com/go-git/go-git/v6/plumbing/format/config" + "github.com/go-git/go-git/v6/plumbing/format/packfile" + "github.com/klauspost/compress/zlib" +) + +// This file is a small git index-pack: it turns a pushed packfile into the +// objects this package stores. Its memory does not grow with the size of the +// history or with the size of any one object. Only a small index grows, by +// about 60 bytes for each object. +// +// It replaces a scratch go-git filesystem.Storage. That storage builds its +// index with a packfile.Parser that has no storage attached, so the parser +// runs outside its low-memory mode and keeps the inflated body of every base +// and every resolved delta until the whole pack is parsed. For golang/go that +// is more than 20 GiB of heap for a 465 MiB pack. +// +// The work is two passes: +// +// 1. scanPack reads the pack once, as it arrives. It stages the bytes in a +// local file, verifies the trailer checksum, and keeps an index entry for +// each object: where it is, how big it is, and which object it is a delta +// against. It inflates every entry to find where the entry ends, and it +// never keeps a body. +// 2. The resolver walks each delta tree depth first, from its base. A body +// lives only while an object below it still needs it as a base, so a +// linear chain costs two bodies at a time. Small bodies stay in memory and +// large ones go to disk; see bodySink. This is how git's own index-pack +// works. + +// incomingObject is one entry of the pushed pack, in pack order. +type incomingObject struct { + offset int64 // the entry header + content int64 // the zlib stream + size int64 // inflated length: the object, or the delta's instruction stream + base int32 // index of the OFS-delta base, or -1 + typ plumbing.ObjectType +} + +// refDelta is a REF-delta entry, found through its base's hash. A client that +// was offered ofs-delta rarely sends one. +type refDelta struct { + base plumbing.Hash + idx int32 + done bool +} + +// incomingPack is the index scanPack builds over the staged pack. +type incomingPack struct { + f io.ReaderAt + end int64 // where the trailer starts + objs []incomingObject + hashSize int + // hashes holds hashSize bytes per object. Only whole objects are filled + // in: a delta's hash is known only once it is resolved, and it is used + // straight away. + hashes []byte + + // kidStart and kids list the OFS-delta children of each object, CSR + // style: the children of i are kids[kidStart[i]:kidStart[i+1]]. + kidStart []int32 + kids []int32 + + refs []refDelta // sorted by base +} + +// maxIncomingPrealloc caps the capacity reserved from the pack header's +// object count, which the client controls. +const maxIncomingPrealloc = 1 << 20 + +// packScanner is the reader under scanPack. It counts and checksums every +// byte it hands out, and it copies each byte to tee, so the staged file holds +// exactly the pack. +// +// It is an io.ByteReader, which is what keeps zlib from reading ahead: flate +// reads a ByteReader one byte at a time and stops at the end of its stream. +// Without that, the scanner would consume bytes past the end of the pack, and +// on a git:// or SSH socket it would block waiting for bytes the client never +// sends, because the client is waiting for report-status. +type packScanner struct { + br *bufio.Reader + n int64 + sum hash.Hash + tee io.Writer + err error // the first tee error + + // pend batches single bytes from ReadByte, which flate calls for nearly + // every byte, so the checksum and the tee see large writes. + pend []byte +} + +func (s *packScanner) flush() { + if len(s.pend) == 0 { + return + } + s.sum.Write(s.pend) + if s.tee != nil && s.err == nil { + _, s.err = s.tee.Write(s.pend) + } + s.pend = s.pend[:0] +} + +func (s *packScanner) ReadByte() (byte, error) { + if s.err != nil { + return 0, s.err + } + b, err := s.br.ReadByte() + if err != nil { + return 0, err + } + s.pend = append(s.pend, b) + s.n++ + if len(s.pend) == cap(s.pend) { + s.flush() + } + return b, nil +} + +func (s *packScanner) Read(p []byte) (int, error) { + s.flush() + if s.err != nil { + return 0, s.err + } + n, err := s.br.Read(p) + if n > 0 { + s.n += int64(n) + s.sum.Write(p[:n]) + if s.tee != nil { + if _, werr := s.tee.Write(p[:n]); werr != nil { + s.err = werr + return n, werr + } + } + } + return n, err +} + +// scanPack reads exactly one packfile from r and indexes it. Every byte it +// reads is copied to tee, when tee is not nil. It stops at the end of the +// trailer and does not read further. +func scanPack(r io.Reader, tee io.Writer, of formatcfg.ObjectFormat, hashSize int) (*incomingPack, error) { + br, ok := r.(*bufio.Reader) + if !ok { + br = bufio.NewReaderSize(r, 64<<10) + } + s := &packScanner{br: br, tee: tee, pend: make([]byte, 0, 32<<10)} + if of == formatcfg.SHA256 { + s.sum = sha256.New() + } else { + s.sum = sha1.New() + } + + var hdr [12]byte + if _, err := io.ReadFull(s, hdr[:]); err != nil { + return nil, fmt.Errorf("read pack header: %w", err) + } + if string(hdr[:4]) != "PACK" { + return nil, fmt.Errorf("%w: bad signature", packfile.ErrMalformedPackfile) + } + if v := binary.BigEndian.Uint32(hdr[4:8]); v != 2 && v != 3 { + return nil, fmt.Errorf("%w: unsupported version %d", packfile.ErrMalformedPackfile, v) + } + count := binary.BigEndian.Uint32(hdr[8:12]) + if count > 1<<31-1 { + return nil, fmt.Errorf("%w: %d objects is too many", packfile.ErrMalformedPackfile, count) + } + + n := min(int(count), maxIncomingPrealloc) + p := &incomingPack{ + hashSize: hashSize, + objs: make([]incomingObject, 0, n), + hashes: make([]byte, 0, n*hashSize), + } + + hs := plumbing.NewHasher(of, plumbing.BlobObject, 0) + ref := make([]byte, hashSize) + for range count { + o, err := scanEntryHeader(s, p, ref) + if err != nil { + return nil, err + } + + idx := int32(len(p.objs)) + p.hashes = append(p.hashes, make([]byte, hashSize)...) + if o.typ == plumbing.REFDeltaObject { + h, _ := plumbing.FromBytes(ref) + p.refs = append(p.refs, refDelta{base: h, idx: idx}) + } + + var dst io.Writer = io.Discard + if !o.typ.IsDelta() { + hs.Reset(o.typ, o.size) + dst = hs + } + if err := inflateTo(s, dst, o); err != nil { + return nil, err + } + if !o.typ.IsDelta() { + copy(p.hashes[int(idx)*hashSize:], hs.Sum().Bytes()) + } + p.objs = append(p.objs, o) + } + + s.flush() + if s.err != nil { + return nil, s.err + } + p.end = s.n + want := s.sum.Sum(nil) + got := make([]byte, len(want)) + if _, err := io.ReadFull(s.br, got); err != nil { + return nil, fmt.Errorf("read pack trailer: %w", err) + } + if tee != nil { + if _, err := tee.Write(got); err != nil { + return nil, err + } + } + if !bytes.Equal(got, want) { + return nil, fmt.Errorf("%w: pack checksum mismatch", packfile.ErrMalformedPackfile) + } + + p.linkKids() + sort.Slice(p.refs, func(i, j int) bool { + return bytes.Compare(p.refs[i].base.Bytes(), p.refs[j].base.Bytes()) < 0 + }) + return p, nil +} + +// scanEntryHeader reads one entry's header: its type and size, then the base +// reference of a delta. A REF-delta's base hash lands in ref. +func scanEntryHeader(s *packScanner, p *incomingPack, ref []byte) (incomingObject, error) { + o := incomingObject{offset: s.n, base: -1} + + c, err := s.ReadByte() + if err != nil { + return o, fmt.Errorf("read entry header at %d: %w", o.offset, err) + } + o.typ = plumbing.ObjectType((c >> 4) & 7) + size := uint64(c & 0x0f) + for shift := uint(4); c&0x80 != 0; shift += 7 { + if shift > 57 { + return o, fmt.Errorf("%w: entry size at %d overflows", packfile.ErrMalformedPackfile, o.offset) + } + if c, err = s.ReadByte(); err != nil { + return o, fmt.Errorf("read entry header at %d: %w", o.offset, err) + } + size |= uint64(c&0x7f) << shift + } + if size > 1<<62 { + return o, fmt.Errorf("%w: entry size at %d overflows", packfile.ErrMalformedPackfile, o.offset) + } + o.size = int64(size) + + switch o.typ { + case plumbing.CommitObject, plumbing.TreeObject, plumbing.BlobObject, plumbing.TagObject: + case plumbing.OFSDeltaObject: + // git's offset encoding adds one before each shift, so every length + // has exactly one encoding. + if c, err = s.ReadByte(); err != nil { + return o, err + } + dist := int64(c & 0x7f) + for c&0x80 != 0 { + if dist > 1<<55 { + return o, fmt.Errorf("%w: ofs-delta distance at %d overflows", packfile.ErrMalformedPackfile, o.offset) + } + if c, err = s.ReadByte(); err != nil { + return o, err + } + dist = (dist+1)<<7 | int64(c&0x7f) + } + if dist <= 0 || dist > o.offset { + return o, fmt.Errorf("%w: invalid ofs-delta distance at %d", packfile.ErrMalformedPackfile, o.offset) + } + // Entries arrive in offset order, so the base is already in objs, and + // a binary search finds it without a map. + at := o.offset - dist + j := sort.Search(len(p.objs), func(k int) bool { return p.objs[k].offset >= at }) + if j == len(p.objs) || p.objs[j].offset != at { + return o, fmt.Errorf("%w: ofs-delta at %d names no entry at %d", packfile.ErrMalformedPackfile, o.offset, at) + } + o.base = int32(j) + case plumbing.REFDeltaObject: + if _, err := io.ReadFull(s, ref); err != nil { + return o, err + } + default: + return o, fmt.Errorf("%w: invalid object type %d at %d", packfile.ErrMalformedPackfile, o.typ, o.offset) + } + + o.content = s.n + return o, nil +} + +// inflateTo inflates entry o from r into dst, and fails unless the stream +// holds exactly the size the header declared. Reading to the zlib EOF also +// consumes the stream's Adler-32 trailer, so r is left at the next entry. +func inflateTo(r io.Reader, dst io.Writer, o incomingObject) error { + zr, err := getInflater(r) + if err != nil { + return fmt.Errorf("inflate entry at %d: %w", o.offset, err) + } + defer zr.Close() + + n, err := io.Copy(dst, io.LimitReader(zr, o.size+1)) + if err != nil { + return fmt.Errorf("inflate entry at %d: %w", o.offset, err) + } + if n != o.size { + return fmt.Errorf("%w: entry at %d inflates to %d bytes or more, header says %d", + packfile.ErrMalformedPackfile, o.offset, n, o.size) + } + return nil +} + +func (p *incomingPack) linkKids() { + p.kidStart = make([]int32, len(p.objs)+1) + for _, o := range p.objs { + if o.base >= 0 { + p.kidStart[o.base+1]++ + } + } + for i := 1; i < len(p.kidStart); i++ { + p.kidStart[i] += p.kidStart[i-1] + } + p.kids = make([]int32, p.kidStart[len(p.objs)]) + fill := make([]int32, len(p.objs)) + for i, o := range p.objs { + if o.base >= 0 { + p.kids[p.kidStart[o.base]+fill[o.base]] = int32(i) + fill[o.base]++ + } + } +} + +func (p *incomingPack) hash(i int32) plumbing.Hash { + h, _ := plumbing.FromBytes(p.hashes[int(i)*p.hashSize : int(i+1)*p.hashSize]) + return h +} + +// children returns every delta built on i, whose hash is h: its OFS-delta +// children, then any REF-delta that names h and is not yet resolved. The REF +// entries are marked done here, so a duplicate base cannot resolve them +// twice. An i of -1 is a base from outside the pack. +func (p *incomingPack) children(i int32, h plumbing.Hash) []int32 { + var out []int32 + if i >= 0 { + out = append(out, p.kids[p.kidStart[i]:p.kidStart[i+1]]...) + } + if len(p.refs) == 0 { + return out + } + k := sort.Search(len(p.refs), func(k int) bool { return p.refs[k].base.Compare(h.Bytes()) >= 0 }) + for ; k < len(p.refs) && p.refs[k].base == h; k++ { + if !p.refs[k].done { + p.refs[k].done = true + out = append(out, p.refs[k].idx) + } + } + return out +} + +// section is the zlib stream of entry i. It ends at the next entry, or at the +// trailer for the last one. +func (p *incomingPack) section(i int32) *io.SectionReader { + end := p.end + if int(i)+1 < len(p.objs) { + end = p.objs[i+1].offset + } + o := p.objs[i] + return io.NewSectionReader(p.f, o.content, end-o.content) +} + +// open streams entry i's inflated body, for a whole object that nothing +// deltas against and that therefore never needs to be anywhere but the +// container. +func (p *incomingPack) open(i int32) (io.ReadCloser, error) { + zr, err := getInflater(p.section(i)) + if err != nil { + return nil, fmt.Errorf("inflate entry at %d: %w", p.objs[i].offset, err) + } + return struct { + io.Reader + io.Closer + }{io.LimitReader(zr, p.objs[i].size), zr}, nil +} + +// inflater is a zlib reader and the buffer under it, pooled because every +// object in a push is inflated at least twice and a fresh reader allocates a +// 32 KiB window each time. +type inflater struct { + br *bufio.Reader + zr io.ReadCloser +} + +var inflaterPool sync.Pool + +// getInflater reads a zlib stream from r. A reader that is an io.ByteReader +// is used as is, so the stream is not read past its end; any other reader +// gets a buffer, which can read ahead. +func getInflater(r io.Reader) (*inflater, error) { + src := r + v, _ := inflaterPool.Get().(*inflater) + if v == nil { + v = &inflater{br: bufio.NewReaderSize(nil, 32<<10)} + } + if _, ok := r.(io.ByteReader); !ok { + v.br.Reset(r) + src = v.br + } + if v.zr == nil { + zr, err := zlib.NewReader(src) + if err != nil { + return nil, err + } + v.zr = zr + return v, nil + } + if err := v.zr.(zlib.Resetter).Reset(src, nil); err != nil { + return nil, err + } + return v, nil +} + +func (in *inflater) Read(p []byte) (int, error) { return in.zr.Read(p) } + +// Close returns the inflater to the pool. It must not be used after. +func (in *inflater) Close() error { + err := in.zr.Close() + in.br.Reset(nil) + inflaterPool.Put(in) + return err +} + +// payloadObject is the plumbing.EncodedObject that packSegment.add copies +// from. The segment needs only a size and a reader. hash is there so its +// error messages name the object. +type payloadObject struct { + hash plumbing.Hash + typ plumbing.ObjectType + size int64 + open func() (io.ReadCloser, error) +} + +func (o *payloadObject) Hash() plumbing.Hash { return o.hash } +func (o *payloadObject) Type() plumbing.ObjectType { return o.typ } +func (o *payloadObject) SetType(t plumbing.ObjectType) { o.typ = t } +func (o *payloadObject) Size() int64 { return o.size } +func (o *payloadObject) SetSize(n int64) { o.size = n } +func (o *payloadObject) Reader() (io.ReadCloser, error) { return o.open() } +func (o *payloadObject) Writer() (io.WriteCloser, error) { + return nil, errors.New("tigris: payload objects are read-only") +} + +// The resolver's memory bound. A body stays in memory when it is at most +// bodyMemLimit and the bodies already in memory leave room for it under +// bodyMemBudget. Everything else goes to a temp file. The budget covers +// every body alive at once, which is what keeps a branching delta tree, or +// a chain of large objects, from adding up. +// +// Variables so a test can lower them; treat them as constants everywhere +// else. +var ( + bodyMemLimit int64 = 8 << 20 // 8 MiB + bodyMemBudget int64 = 64 << 20 // 64 MiB +) + +// body is an object, or a delta, while the resolver needs it: in memory, or +// in a temp file. +type body struct { + mem []byte + f *os.File + size int64 +} + +func (b *body) readerAt() io.ReaderAt { + if b.f != nil { + return b.f + } + return bytes.NewReader(b.mem) +} + +func (b *body) open() (io.ReadCloser, error) { + return io.NopCloser(io.NewSectionReader(b.readerAt(), 0, b.size)), nil +} + +// bodySink collects one body. It starts in memory and moves to a temp file +// as soon as the body passes what the memory bound admits. It never trusts a +// size hint, because every hint comes from the client. +type bodySink struct { + r *resolver + limit int64 // the most this body may hold in memory + buf []byte + f *os.File + w *bufio.Writer + n int64 +} + +func (r *resolver) newSink(hint int64) *bodySink { + limit := min(bodyMemLimit, bodyMemBudget-r.mem) + s := &bodySink{r: r, limit: limit} + if hint <= limit { + s.buf = make([]byte, 0, max(hint, 0)) + } + return s +} + +func (s *bodySink) Write(p []byte) (int, error) { + if s.f == nil && int64(len(s.buf)+len(p)) > s.limit { + f, err := os.CreateTemp(s.r.dir, "body-*") + if err != nil { + return 0, fmt.Errorf("spill object body: %w", err) + } + s.f = f + s.w = bufio.NewWriterSize(f, 256<<10) + if _, err := s.w.Write(s.buf); err != nil { + return 0, err + } + s.buf = nil + } + s.n += int64(len(p)) + if s.f != nil { + return s.w.Write(p) + } + s.buf = append(s.buf, p...) + return len(p), nil +} + +func (s *bodySink) finish() (*body, error) { + if s.f != nil { + if err := s.w.Flush(); err != nil { + return nil, fmt.Errorf("spill object body: %w", err) + } + return &body{f: s.f, size: s.n}, nil + } + s.r.mem += int64(len(s.buf)) + return &body{mem: s.buf, size: s.n}, nil +} + +// release frees a body. It is safe on nil and on a body already released. +func (r *resolver) release(b *body) { + if b == nil { + return + } + if b.f != nil { + b.f.Close() + os.Remove(b.f.Name()) + b.f = nil + } + if b.mem != nil { + r.mem -= int64(len(b.mem)) + b.mem = nil + } +} + +// external resolves the base of a REF-delta that the pack does not carry. A +// client should not send one, because receive-pack advertises no-thin, but +// the base can still be in the repository already. +type externalBase func(plumbing.Hash) (plumbing.EncodedObject, error) + +// resolver writes a scanned pack into containers, base before delta. +type resolver struct { + w *packWriter + p *incomingPack + of formatcfg.ObjectFormat + external externalBase + dir string // the temp directory spilled bodies go to + + mem int64 // bytes of body held in memory right now + + byteLimit int64 + seg *packSegment + // serial numbers the container being built. An object's delta form is + // legal only when its base went into the same serial. + serial int32 +} + +// run resolves and writes every object in the pack. +func (r *resolver) run() error { + p := r.p + for i := range p.objs { + if p.objs[i].typ.IsDelta() { + continue + } + if err := r.root(int32(i)); err != nil { + return err + } + } + + // Any REF-delta still pending names a base outside the pack. + for k := range p.refs { + if p.refs[k].done { + continue + } + h := p.refs[k].base + if r.external == nil { + return fmt.Errorf("%w: ref-delta base %s is not in the pack", plumbing.ErrObjectNotFound, h) + } + obj, err := r.external(h) + if err != nil { + return fmt.Errorf("resolve ref-delta base %s: %w", h, err) + } + b, err := r.readBody(obj) + if err != nil { + return fmt.Errorf("resolve ref-delta base %s: %w", h, err) + } + // The base is not in any container this push builds, so serial -1 + // keeps every child in its whole form. + if err := r.walk(p.children(-1, h), h, obj.Type(), b, -1, 0, 1); err != nil { + return err + } + } + return nil +} + +func (r *resolver) readBody(obj plumbing.EncodedObject) (*body, error) { + rd, err := obj.Reader() + if err != nil { + return nil, err + } + defer rd.Close() + sink := r.newSink(obj.Size()) + if _, err := io.Copy(sink, rd); err != nil { + return nil, err + } + return sink.finish() +} + +// inflateBody reads entry i's whole inflated body into a body. +func (r *resolver) inflateBody(i int32) (*body, error) { + o := r.p.objs[i] + sink := r.newSink(o.size) + if err := inflateTo(r.p.section(i), sink, o); err != nil { + return nil, err + } + return sink.finish() +} + +func (r *resolver) root(i int32) error { + p := r.p + o := p.objs[i] + h := p.hash(i) + kids := p.children(i, h) + + if len(kids) == 0 { + // Nothing needs this body as a base, so it streams from the pack file + // into the container. + so := storedObject{ + payload: &payloadObject{hash: h, typ: o.typ, size: o.size, open: func() (io.ReadCloser, error) { return p.open(i) }}, + hash: h, typ: o.typ, raw: o.size, + } + _, _, err := r.place(so, nil, plumbing.ZeroHash, -1, 0) + return err + } + + b, err := r.inflateBody(i) + if err != nil { + return err + } + so := storedObject{ + payload: &payloadObject{hash: h, typ: o.typ, size: b.size, open: b.open}, + hash: h, typ: o.typ, raw: o.size, + } + serial, _, err := r.place(so, nil, plumbing.ZeroHash, -1, 0) + if err != nil { + r.release(b) + return err + } + return r.walk(kids, h, o.typ, b, serial, 0, 1) +} + +// walk resolves kids, which are all deltas against base, and then everything +// built on each of them. It owns base and releases it. +// +// baseChain is how many stored delta links sit under the base: 0 when the +// base is stored whole. depth is the client's chain depth, which can be +// deeper, because place stores some objects whole. +func (r *resolver) walk(kids []int32, baseHash plumbing.Hash, typ plumbing.ObjectType, base *body, baseSerial int32, baseChain, depth int) error { + defer r.release(base) + if depth > maxIncomingDeltaDepth { + return fmt.Errorf("%w: delta chain deeper than %d", packfile.ErrMalformedPackfile, maxIncomingDeltaDepth) + } + + for n, c := range kids { + delta, err := r.inflateBody(c) + if err != nil { + return err + } + out, h, err := r.apply(typ, base, delta) + if err != nil { + r.release(delta) + return fmt.Errorf("apply delta at %d: %w", r.p.objs[c].offset, err) + } + if n == len(kids)-1 { + // The last child is the base's last use. Free it before going + // deeper, so a linear chain holds two bodies and not the chain. + r.release(base) + } + + so := storedObject{ + payload: &payloadObject{hash: h, typ: typ, size: out.size, open: out.open}, + hash: h, typ: typ, raw: out.size, + } + serial, chain, err := r.place(so, delta, baseHash, baseSerial, baseChain) + r.release(delta) + if err != nil { + r.release(out) + return err + } + + grand := r.p.children(c, h) + if len(grand) == 0 { + r.release(out) + continue + } + if err := r.walk(grand, h, typ, out, serial, chain, depth+1); err != nil { + return err + } + } + return nil +} + +// maxIncomingDeltaDepth matches git's own ceiling on a delta chain, +// (1 << OE_DEPTH_BITS) - 1 in pack-objects. +const maxIncomingDeltaDepth = 4095 + +// place adds one object to the container being built, sealing it first when +// the object would push it past the byte cap. It returns the serial of the +// container the object went into, and the number of stored delta links under +// it: 0 when it is stored whole. +// +// delta is the client's delta against baseHash, or nil. It is stored instead +// of the whole object only when all three hold: +// +// - the base went into this same container, so a read never chases a base +// into another container; +// - it is smaller than the object, so a read never pays a base fetch for no +// bytes back; +// - the stored chain stays within maxDeltaDepth, the longest chain a read +// walks. A client can send chains up to 4095 links. +func (r *resolver) place(so storedObject, delta *body, baseHash plumbing.Hash, baseSerial int32, baseChain int) (int32, int, error) { + // The cap seals before the add: a container at 127 MiB must not swallow + // a 500 MiB blob. The len(recs) guard lets an object bigger than the + // whole cap have a container to itself. + // + // seg.offset counts stored bytes and so.raw is the raw size. The codec + // policy guarantees stored <= raw, so so.raw is a safe upper bound. + if r.seg != nil && len(r.seg.recs) > 0 && r.seg.offset+so.raw > r.byteLimit { + full := r.seg + r.seg = nil // ownership moves to seal + if err := r.w.seal(full); err != nil { + return 0, 0, err + } + } + if r.seg == nil { + seg, err := newPackSegment(r.w.s) + if err != nil { + return 0, 0, err + } + r.seg = seg + r.serial++ + } + + chain := 0 + if delta != nil && baseSerial == r.serial && delta.size < so.raw && baseChain < maxDeltaDepth { + so.payload = &payloadObject{hash: so.hash, typ: plumbing.REFDeltaObject, size: delta.size, open: delta.open} + so.base = baseHash + chain = baseChain + 1 + } + if err := r.seg.add(so); err != nil { + return 0, 0, err + } + return r.serial, chain, nil +} + +// apply rebuilds an object of type typ from base and a git delta, and hashes +// it on the way. It reads base through io.ReaderAt, so a base on disk is never +// loaded whole. +// +// The format is git's: the source and target sizes as little-endian base-128 +// numbers, then commands. A command with the high bit set copies a range of +// the base; one without it inserts that many literal bytes from the delta. +func (r *resolver) apply(typ plumbing.ObjectType, base, delta *body) (*body, plumbing.Hash, error) { + drc, _ := delta.open() + defer drc.Close() + d := bufio.NewReaderSize(drc, 64<<10) + + src, err := readDeltaSize(d) + if err != nil { + return nil, plumbing.ZeroHash, err + } + if int64(src) != base.size { + return nil, plumbing.ZeroHash, packfile.ErrInvalidDelta + } + tgt, err := readDeltaSize(d) + if err != nil { + return nil, plumbing.ZeroHash, err + } + if tgt > 1<<62 { + return nil, plumbing.ZeroHash, packfile.ErrInvalidDelta + } + + sink := r.newSink(int64(tgt)) + hs := plumbing.NewHasher(r.of, typ, int64(tgt)) + out := bufio.NewWriterSize(io.MultiWriter(sink, hs), 64<<10) + fail := func(err error) (*body, plumbing.Hash, error) { + if b, ferr := sink.finish(); ferr == nil { + r.release(b) + } + return nil, plumbing.ZeroHash, err + } + + ra := base.readerAt() + remaining := tgt + for remaining > 0 { + cmd, err := d.ReadByte() + if err != nil { + return fail(packfile.ErrInvalidDelta) + } + switch { + case cmd&0x80 != 0: + var off, sz uint64 + for k := range 4 { + if cmd&(1< remaining || off+sz > src { + return fail(packfile.ErrInvalidDelta) + } + if _, err := io.Copy(out, io.NewSectionReader(ra, int64(off), int64(sz))); err != nil { + return fail(err) + } + remaining -= sz + case cmd != 0: + sz := uint64(cmd) + if sz > remaining { + return fail(packfile.ErrInvalidDelta) + } + if _, err := io.CopyN(out, d, int64(sz)); err != nil { + return fail(packfile.ErrInvalidDelta) + } + remaining -= sz + default: + return fail(packfile.ErrDeltaCmd) + } + } + // Every byte of the delta must be used, as git requires. + if _, err := d.ReadByte(); err != io.EOF { + return fail(packfile.ErrInvalidDelta) + } + if err := out.Flush(); err != nil { + return fail(err) + } + b, err := sink.finish() + if err != nil { + return nil, plumbing.ZeroHash, err + } + return b, hs.Sum(), nil +} + +// readDeltaSize reads one of a delta header's two sizes. +func readDeltaSize(d io.ByteReader) (uint64, error) { + var n uint64 + for shift := uint(0); ; shift += 7 { + if shift > 63 { + return 0, packfile.ErrInvalidDelta + } + b, err := d.ReadByte() + if err != nil { + return 0, packfile.ErrInvalidDelta + } + n |= uint64(b&0x7f) << shift + if b&0x80 == 0 { + return n, nil + } + } +} diff --git a/internal/storage/tigris/packindex.go b/internal/storage/tigris/packindex.go index d88b5da..729e2d0 100644 --- a/internal/storage/tigris/packindex.go +++ b/internal/storage/tigris/packindex.go @@ -848,8 +848,9 @@ func (s *Storer) packPayload(h plumbing.Hash, e packEntry) ([]byte, error) { } // maxDeltaDepth bounds how far a read will walk a delta chain before giving up. -// It matches go-git's own limit (packfile.maxDepth), which is what bounds the -// chains a push can hand us in the first place. A .cue is just bytes in a +// A read admits a chain of exactly this many links. It matches git's default +// pack.depth, but a client can send chains up to 4095 links, so the writer +// (resolver.place in indexpack.go) is what holds stored chains to this bound. A .cue is just bytes in a // bucket, so this is also the guard that keeps a corrupt or hostile index from // recursing without end; the visited set beside it catches short cycles that // would otherwise spin until the depth ran out. @@ -908,7 +909,10 @@ func (s *Storer) framePacked(h plumbing.Hash, e packEntry, payload []byte, depth // something it did not keep. Reporting ErrObjectNotFound here would let a // damaged index read as "this object was never pushed". func (s *Storer) deltaBase(base plumbing.Hash, depth int) (plumbing.EncodedObject, error) { - if depth >= maxDeltaDepth { + // depth is the link that reaches this base, so a chain of exactly + // maxDeltaDepth links fetches its whole object at depth maxDeltaDepth. + // framePacked refuses the link past it. + if depth > maxDeltaDepth { return nil, fmt.Errorf("%w: delta chain through %s is deeper than %d links", errBadCue, base.String(), maxDeltaDepth) } diff --git a/internal/storage/tigris/packwriter.go b/internal/storage/tigris/packwriter.go index 63e6f28..6ccd228 100644 --- a/internal/storage/tigris/packwriter.go +++ b/internal/storage/tigris/packwriter.go @@ -1,6 +1,7 @@ package tigris import ( + "bufio" "bytes" "context" "crypto/sha256" @@ -14,67 +15,92 @@ import ( "time" "github.com/aws/aws-sdk-go-v2/service/s3" - "github.com/go-git/go-billy/v6/osfs" "github.com/go-git/go-git/v6/plumbing" - "github.com/go-git/go-git/v6/plumbing/cache" - "github.com/go-git/go-git/v6/storage/filesystem" "github.com/klauspost/compress/zstd" - "golang.org/x/sync/errgroup" ) -// PackfileWriter accepts an incoming git packfile — already exactly -// delimited by the caller (see cmd/objgitd/receivepack.go's writePack, -// which needs no changes: its own st.(storer.PackfileWriter) type assertion -// just starts succeeding once this method exists). Rather than storing git's -// own delta-compressed pack format, it decodes the pack (by handing it to a -// scratch go-git filesystem.Storage, which does all the actual pack/delta -// work — see docs/reference/tigris-backend.md) and re-encodes every object, -// fully resolved, into this package's own flat bin/cue container. +// PackfileWriter accepts an incoming git packfile, already delimited by the +// caller (see writePack in cmd/objgitd/receivepack.go). It does not store +// git's own delta-compressed pack format. Write stages the pack in a local +// temp file, and Close resolves every object and re-encodes it into this +// package's own flat bin/cue containers. See indexpack.go for how. func (s *Storer) PackfileWriter() (io.WriteCloser, error) { - dir, err := os.MkdirTemp("", "objgit-tigris-scratch-*") + f, err := os.CreateTemp("", "objgit-tigris-incoming-*") if err != nil { - return nil, fmt.Errorf("tigris: create scratch dir: %w", err) + return nil, fmt.Errorf("tigris: create incoming pack staging file: %w", err) } - - // A real temp dir, not memfs: incoming packs can be large. - scratch := filesystem.NewStorageWithOptions( - osfs.New(dir), - cache.NewObjectLRUDefault(), - filesystem.Options{ObjectFormat: s.of}, - ) - inner, err := scratch.PackfileWriter() - if err != nil { - os.RemoveAll(dir) - return nil, fmt.Errorf("tigris: open scratch packfile writer: %w", err) - } - return &packWriter{s: s, dir: dir, scratch: scratch, inner: inner}, nil + return &packWriter{s: s, f: f, buf: bufio.NewWriterSize(f, 1<<20)}, nil } type packWriter struct { - s *Storer - dir string - scratch *filesystem.Storage - inner io.WriteCloser - done bool + s *Storer + f *os.File + buf *bufio.Writer + n int64 + pack *incomingPack // set by ReadPack, which indexes as it stages + bad bool // ReadPack failed, so Close only cleans up + done bool } -func (w *packWriter) Write(p []byte) (int, error) { return w.inner.Write(p) } +func (w *packWriter) Write(p []byte) (int, error) { + n, err := w.buf.Write(p) + w.n += int64(n) + return n, err +} + +// ReadPack reads exactly one packfile from r, and stops at its trailer. It +// stages the pack and indexes it in one pass, so Close does not read the +// staged file a second time. It never holds an object body in memory. +// +// cmd/objgitd's writePack prefers this to Write. The pack must end on its +// own framing there, because a git:// or SSH client keeps the connection open +// after the pack while it waits for report-status. +func (w *packWriter) ReadPack(r io.Reader) error { + cw := &countingWriter{w: w.buf} + p, err := scanPack(r, cw, w.s.of, w.s.oh.Size()) + w.n += cw.n + if err != nil { + w.bad = true + return fmt.Errorf("tigris: decode incoming pack: %w", err) + } + w.pack = p + return nil +} func (w *packWriter) Close() error { if w.done { return nil } w.done = true - defer os.RemoveAll(w.dir) + defer os.Remove(w.f.Name()) + defer w.f.Close() - if err := w.inner.Close(); err != nil { - return fmt.Errorf("tigris: decode incoming pack: %w", err) + if w.bad { + return nil + } + if err := w.buf.Flush(); err != nil { + return fmt.Errorf("tigris: stage incoming pack: %w", err) + } + if w.n == 0 { + return nil // defensive: writePack only calls us when a packfile is expected } - objs, order, err := w.planObjects() + pack := w.pack + if pack == nil { + // The pack came through Write, so index the staged copy now. + var err error + pack, err = scanPack(io.NewSectionReader(w.f, 0, w.n), nil, w.s.of, w.s.oh.Size()) + if err != nil { + return fmt.Errorf("tigris: decode incoming pack: %w", err) + } + } + pack.f = w.f + + dir, err := os.MkdirTemp("", "objgit-tigris-bodies-*") if err != nil { - return err + return fmt.Errorf("tigris: create body spill dir: %w", err) } + defer os.RemoveAll(dir) // One push becomes as many containers as its total size needs: the walk // seals a segment the moment it is full and opens the next one lazily, so a @@ -91,100 +117,30 @@ func (w *packWriter) Close() error { byteLimit = maxPackBytes } - var seg *packSegment + r := &resolver{w: w, p: pack, of: w.s.of, dir: dir, byteLimit: byteLimit, external: w.externalBase} defer func() { - if seg != nil { // only reachable on an error before this segment sealed - seg.discard() + if r.seg != nil { // only reachable on an error before this segment sealed + r.seg.discard() } }() - - // inSeg is the containment rule made concrete: a delta may only be stored - // as a delta when its base is in the container being built. It resets on - // every seal, so a delta stranded on the far side of a split is demoted to - // its whole form rather than pointing at a sibling container. - // - // No size hint: only the byte cap bounds a container now, so nothing here - // knows how many objects one holds. clear keeps the buckets, so the map - // grows once to the largest container's object count and stays there. - inSeg := map[plumbing.Hash]struct{}{} - - walkErr := func() error { - for _, idx := range order { - p := objs[idx] - so, err := w.payloadFor(p, inSeg) - if err != nil { - return err - } - - // The cap seals *before* the add: a container sitting at 127 MiB - // must not swallow a 500 MiB blob and land at 627 MiB. The - // len(seg.recs) > 0 guard is what permits the one legal spill — an - // object larger than the whole cap gets a container to itself, - // since it has to live somewhere. That container then seals on the - // next object's check here, or at the end of the walk below. - // - // p.size is trustworthy: add fails the push if the bytes it copies - // disagree with the payload's own declared size. - // - // seg.offset counts *stored* bytes while p.size is the raw size, so - // the comparison mixes the two. It stays correct because the codec - // policy guarantees stored <= raw (compress.go, and the rewind in - // writeProbed), which makes p.size a valid upper bound on what this - // object can add — more so for a delta, whose payload is smaller - // still. The only cost is sealing a little early. - if seg != nil && len(seg.recs) > 0 && seg.offset+p.size > byteLimit { - full := seg - seg = nil // ownership moves to seal, which owns its staging files - if err := w.seal(full); err != nil { - return err - } - clear(inSeg) - - // The seal moved the goalposts: this object's base is no longer - // in the container being built, so its delta form is no longer - // legal. Ask again, now that inSeg is empty. - if so, err = w.payloadFor(p, inSeg); err != nil { - return err - } - } - if seg == nil { - var err error - if seg, err = newPackSegment(w.s); err != nil { - return err - } - } - if err := seg.add(so); err != nil { - return err - } - inSeg[p.hash] = struct{}{} - } - return nil - }() - if walkErr != nil { + if err := r.run(); err != nil { // Any segment this push already sealed is enqueued and cannot be // recalled. Each one is a complete, self-consistent container holding // real objects, so nothing dangles; the error fails the push, so no ref // ever points into the incomplete set. - return fmt.Errorf("tigris: build pack container: %w", walkErr) + return fmt.Errorf("tigris: build pack container: %w", err) } - if seg == nil { - return nil // defensive: writePack only calls us when a packfile is expected + if r.seg == nil { + return nil } - last := seg - seg = nil + last := r.seg + r.seg = nil return w.seal(last) } -// plannedObject is one object's metadata, gathered before any bytes are -// written. Type and size come from the *resolved* object, never from the delta -// header: git records a delta entry's size as the length of its instruction -// stream, not of the object it rebuilds, and every record's raw field means the -// latter. -type plannedObject struct { - hash plumbing.Hash - typ plumbing.ObjectType - size int64 // the reconstructed object's size - base plumbing.Hash // zero when the client did not send this as a delta +// externalBase reads a REF-delta base that is already in the repository. +func (w *packWriter) externalBase(h plumbing.Hash) (plumbing.EncodedObject, error) { + return w.s.EncodedObject(plumbing.AnyObject, h) } // storedObject is one object as a segment will record it: the payload bytes to @@ -199,185 +155,6 @@ type storedObject struct { base plumbing.Hash // zero when payload is the whole object } -// planObjects walks the scratch storage once and returns every object -// (flat) alongside an order — indices into flat — where a delta always -// follows its base. -// -// The ordering is the reason this pass exists. IterEncodedObjects yields in -// index order, which is hash order, so a delta can arrive long before the -// object it is built from — and the containment rule cannot be checked against -// a container that has not been filled yet. -// -// order and placed hold indices/flags rather than copies of plannedObject or -// a second map keyed by plumbing.Hash: a 32-byte SHA256 hash makes a -// map[plumbing.Hash]* expensive per entry, and a push the size of a large -// repository's full history has enough objects that the difference is the -// gap between this fitting in memory and not. flat is already hash-sorted -// (the property above), so base lookups use binary search instead of a -// second map. -func (w *packWriter) planObjects() ([]plannedObject, []int32, error) { - iter, err := w.scratch.IterEncodedObjects(plumbing.AnyObject) - if err != nil { - return nil, nil, fmt.Errorf("tigris: walk scratch objects: %w", err) - } - defer iter.Close() - - // Hash, type, and size only: IterEncodedObjects' iterator, like the rest - // of filesystem.ObjectStorage's packed-object path, is documented as not - // safe for concurrent use, so this walk stays on one goroutine. Delta-base - // resolution below is the expensive part and is where the concurrency goes. - var flat []plannedObject - if err := iter.ForEach(func(obj plumbing.EncodedObject) error { - flat = append(flat, plannedObject{hash: obj.Hash(), typ: obj.Type(), size: obj.Size()}) - return nil - }); err != nil { - return nil, nil, fmt.Errorf("tigris: plan pack container: %w", err) - } - - if err := w.resolveDeltaBases(flat); err != nil { - return nil, nil, fmt.Errorf("tigris: plan pack container: %w", err) - } - - // Emit bases first. Marking before recursing is what makes a cycle - // terminate: a chain that loops back on itself simply places the object it - // looped to, and the delta behind it is demoted when its base turns out not - // to be in the container yet. - order := make([]int32, 0, len(flat)) - placed := make([]bool, len(flat)) - var emit func(i int32) - emit = func(i int32) { - if placed[i] { - return - } - placed[i] = true - if base := flat[i].base; base != plumbing.ZeroHash { - if j, ok := findByHash(flat, base); ok { - emit(j) - } - } - order = append(order, i) - } - for i := range flat { - emit(int32(i)) - } - return flat, order, nil -} - -// findByHash locates h in flat by binary search. Callers hold flat in the -// hash order IterEncodedObjects produced, which is the order this depends on. -func findByHash(flat []plannedObject, h plumbing.Hash) (int32, bool) { - i := sort.Search(len(flat), func(i int) bool { - return bytes.Compare(flat[i].hash.Bytes(), h.Bytes()) >= 0 - }) - if i < len(flat) && flat[i].hash == h { - return int32(i), true - } - return -1, false -} - -// deltaScanWorkers bounds how many independent scratch storage handles -// resolveDeltaBases opens at once. Capped rather than left at GOMAXPROCS so a -// huge push does not open more file descriptors against the scratch -// directory than is sensible; the scan is I/O-bound, not CPU-bound, so there -// is little to gain past a handful of workers anyway. -const deltaScanWorkers = 8 - -// resolveDeltaBases fills in flat[i].base for every object the client sent -// as a delta, by asking the scratch storage whether it has one. -// -// filesystem.ObjectStorage's packed-object delta lookup (which this needs) -// reads through Packfile.Scanner(), a single shared, unsynchronized scanner -// over one shared file handle — go-git's own doc comment on Scanner marks it -// not thread-safe. So this cannot fan goroutines out against w.scratch -// itself. Instead each worker opens its own *filesystem.Storage over the -// same on-disk scratch directory: lazy init means each gets an independent -// packfile.Packfile, Scanner, and file descriptor, so there is no shared -// mutable state to race on. flat is partitioned by index range and each -// worker only ever writes the indices in its own range, so flat itself needs -// no synchronization either — distinct slice elements are distinct memory, -// concurrent writes to them are not a race. -// -// Each worker's storage gets a zero-size object cache: every hash here is -// looked up exactly once, so the cache would only hold memory without ever -// serving a hit. -func (w *packWriter) resolveDeltaBases(flat []plannedObject) error { - if len(flat) == 0 { - return nil - } - - workers := min(deltaScanWorkers, len(flat)) - chunk := (len(flat) + workers - 1) / workers - - var g errgroup.Group - for start := 0; start < len(flat); start += chunk { - end := min(start+chunk, len(flat)) - g.Go(func() error { - store := filesystem.NewStorageWithOptions( - osfs.New(w.dir), cache.NewObjectLRU(0), - filesystem.Options{ObjectFormat: w.s.of}, - ) - for i := start; i < end; i++ { - if do, ok := deltaFormOn(store, flat[i].hash); ok { - flat[i].base = do.BaseHash() - } - } - return nil - }) - } - return g.Wait() -} - -// deltaForm asks the scratch storage for h's delta, if the client sent one. -// The scratch storage is a real go-git filesystem.Storage over the pushed -// packfile, so this is the client's own delta chain, not one we computed. -func (w *packWriter) deltaForm(h plumbing.Hash) (plumbing.DeltaObject, bool) { - return deltaFormOn(w.scratch, h) -} - -// deltaFormOn is deltaForm against an explicit storage rather than w.scratch, -// so resolveDeltaBases' workers can each use their own independent instance. -func deltaFormOn(store *filesystem.Storage, h plumbing.Hash) (plumbing.DeltaObject, bool) { - obj, err := store.DeltaObject(plumbing.AnyObject, h) - if err != nil { - // Not fatal: a base we cannot read as a delta is simply stored whole. - return nil, false - } - do, ok := obj.(plumbing.DeltaObject) - if !ok || do.BaseHash() == plumbing.ZeroHash { - return nil, false - } - return do, true -} - -// payloadFor fetches the bytes to store for p: the client's delta when its base -// is already in the container being built, and the whole object otherwise. -func (w *packWriter) payloadFor(p plannedObject, inSeg map[plumbing.Hash]struct{}) (storedObject, error) { - whole := func() (storedObject, error) { - obj, err := w.scratch.EncodedObject(plumbing.AnyObject, p.hash) - if err != nil { - return storedObject{}, fmt.Errorf("tigris: read %s from scratch: %w", p.hash, err) - } - return storedObject{payload: obj, hash: p.hash, typ: p.typ, raw: p.size}, nil - } - - if p.base == plumbing.ZeroHash { - return whole() - } - if _, ok := inSeg[p.base]; !ok { - return whole() - } - do, ok := w.deltaForm(p.hash) - if !ok || do.BaseHash() != p.base { - return whole() - } - // A delta that does not actually save anything is worse than the object: - // it costs a base read on every future fetch for no bytes back. - if do.Size() >= p.size { - return whole() - } - return storedObject{payload: do, hash: p.hash, typ: p.typ, raw: p.size, base: p.base}, nil -} - // packSegment is one container under construction: the staging .bin and the // cue records for the objects written into it so far. // From db4bd4700ed29e212f38cc538dadb6986715f85c Mon Sep 17 00:00:00 2001 From: Xe Iaso Date: Fri, 25 Sep 2026 16:28:55 -0400 Subject: [PATCH 2/8] docs: plan bounded push memory and describe the new pack writer Record the measurements behind the new pack writer, the delta chain depth fault it found, and the next phase: a compact pack index to replace the per-object entries map. Assisted-by: Claude Opus 5.5 via Claude Code Signed-off-by: Xe Iaso --- docs/architecture/tigris-storer.md | 47 +++++--- docs/plans/push-memory.md | 182 +++++++++++++++++++++++++++++ docs/reference/tigris-backend.md | 35 +++--- 3 files changed, 234 insertions(+), 30 deletions(-) create mode 100644 docs/plans/push-memory.md diff --git a/docs/architecture/tigris-storer.md b/docs/architecture/tigris-storer.md index 5906384..e8228c9 100644 --- a/docs/architecture/tigris-storer.md +++ b/docs/architecture/tigris-storer.md @@ -339,19 +339,36 @@ caps below set how much one container holds. It does not store the git pack format. Git packs use delta compression, so one read would have to resolve a chain of deltas. -`PackfileWriter` instead writes the incoming pack to a scratch -`storage/filesystem.Storage` on a local temp directory. That storage decodes -the pack for us, so this package holds no pack-parsing code. - -The writer makes two passes. The first reads the hash, type, size, and delta -base of every object, then orders them so a base always precedes the delta -that needs it — `IterEncodedObjects` returns hash order, which gives no such -guarantee. The second pass copies each payload into a flat `packs/.bin` -and adds one record to `packs/.cue`. A record holds the hash, the type, -the payload codec, the offset, the stored length, the raw size, and the base -hash. `` is the hex SHA-256 of the `.bin`, so the name doubles as a -checksum. Both files upload through the same uploader as loose objects, so the -existing `SetReference` flush covers them. +`PackfileWriter` decodes the pack itself, in `indexpack.go`. That file is a +small version of `git index-pack`. Its memory does not grow with the size of +the history or with the size of one object. See +[the plan](../plans/push-memory.md) for the measurements. + +The writer makes two passes: + +1. `scanPack` reads the pack as it arrives. It stages the bytes in a temp + file, verifies the trailer checksum, and keeps a small index entry for each + object: the offset, the size, the type, and the base of a delta. It keeps + no body. +2. The resolver walks each delta tree depth first, from its base, so a base + always goes into a container before its deltas. It copies each payload into + a flat `packs/.bin` and adds one record to `packs/.cue`. + +A record holds the hash, the type, the payload codec, the offset, the stored +length, the raw size, and the base hash. `` is the hex SHA-256 of the +`.bin`, so the name is also a checksum. Both files upload through the same +uploader as loose objects, so the existing `SetReference` flush covers them. + +A body stays in memory only while a delta below it needs it as a base. A +body goes to a temp file when it is larger than `bodyMemLimit` (8 MiB), or when +the bodies in memory would pass `bodyMemBudget` (64 MiB). The delta applier +reads its base through `io.ReaderAt`, so a base on disk is never loaded whole. + +`cmd/objgitd`'s `writePack` calls `ReadPack` when the writer has it. `ReadPack` +finds the end of the pack in the same pass that stages it. `packScanner` is an +`io.ByteReader`, so zlib never reads past the pack. On git:// and SSH the +client keeps the connection open after the pack, so a read past the pack +blocks. A payload is the object, or the delta the pushing client already computed for it. Keeping that delta is what stops every later clone from deriving one @@ -359,6 +376,10 @@ again: `Storer` implements `storer.DeltaObjectStorer`, so go-git's packer reuses the stored delta instead of running its rolling-hash search. See [the backend reference](../reference/tigris-backend.md#how-a-read-rebuilds-a-delta). +A read walks at most `maxDeltaDepth` (50) delta links, and a client can send +chains of up to 4095 links. The writer therefore stores an object whole when +its stored chain is already 50 links long. + A delta is only kept when its base lands in the same container. When a container seals between the two, the writer stores the whole object instead. That costs a few deltas at each boundary and buys back the rule that every diff --git a/docs/plans/push-memory.md b/docs/plans/push-memory.md new file mode 100644 index 0000000..1a70d0c --- /dev/null +++ b/docs/plans/push-memory.md @@ -0,0 +1,182 @@ +# Plan: bounded memory for large pushes + +## Goal + +A push of any repository, of any size, must complete in bounded memory. The +memory that one push uses must not grow with the size of the history, and must +not grow with the size of one object. + +## The problem + +A push of golang/go (465 MiB pack, 708,000 objects) used more than 21 GiB of +heap. On a laptop it filled swap and did not complete. A push of Xe/x (50 MB +pack, 21,000 objects) used 516 MiB. + +Almost all of that memory was in go-git, and not in this repository. + +`PackfileWriter` gave the pushed pack to a scratch go-git +`filesystem.Storage`. That storage indexes the pack with a `packfile.Parser` +that has no storage attached. Without a storage, the parser does not use its +low-memory mode. It keeps the inflated body of every base object and every +resolved delta until the whole pack is parsed. Memory therefore grew with the +uncompressed size of the full history. + +Three smaller costs came after the parse: + +- The plan pass called `IterEncodedObjects`, which loads each full object to + read its hash, type, and size. +- Eight workers opened eight more copies of the scratch index to find delta + bases. +- The container walk resolved each object again through the go-git packfile + reader. + +A fourth cost was in `cmd/objgitd/receivepack.go`. `writePack` ran a go-git +`Scanner` over the socket to find the end of the pack. That reader cannot +seek, so the scanner buffered the full inflated body of each object. One large +blob therefore needed its full size in memory. + +## Phase 1: replace the scratch storage (done) + +`internal/storage/tigris/indexpack.go` is a small version of `git index-pack`. +It does the work in two passes. + +1. `scanPack` reads the pack once, as it arrives. It stages the bytes in a + temp file and verifies the trailer checksum. For each object it keeps about + 60 bytes: the offset, the size, the type, and the base of a delta. It + inflates each entry to find where the entry ends, but it keeps no body. +2. The resolver walks each delta tree depth first, from its base. A body lives + only while a delta below it needs it as a base. A linear chain therefore + holds two bodies at a time. Each object is inflated, resolved, and hashed + once. + +A body stays in memory when it is 8 MiB or less, and when all bodies in memory +stay under 64 MiB. Other bodies go to a temp directory for the push. The delta +applier reads its base through `io.ReaderAt`, so it never loads a base on disk +into memory. + +`writePack` now hands the stream to the storer through an optional +`ReadPack(io.Reader) error` method. The storer finds the end of the pack in the +same pass that stages and indexes it. `packScanner` is an `io.ByteReader`, so +zlib does not read past the end of the pack. This matters on git:// and SSH, +where the client keeps the connection open after the pack. + +### Results of phase 1 + +The numbers come from a local harness. It feeds a real pack to +`PackfileWriter` over an S3 fake that discards bodies. + +| Push | Before | After | +| ---------------------------- | ------------------------ | ------------------------- | +| golang/go, peak heap | more than 21 GiB | 521 MiB | +| golang/go, time | did not complete | 23 s | +| Xe/x, peak heap | 516 MiB | about 165 MiB | +| Xe/x, time | 14.7 s | 1.3 s | +| 266 MB blob and one delta | both bodies in memory | 35 MiB peak heap | + +A second test reads every object back through a new `Storer` and verifies +its hash. It passes for all 708,000 objects of golang/go. It also passes for a +chain of 50 links, a chain of 197 links, and the 266 MB blob. Each case passes +with the default memory bound, and again with every body sent to disk. + +The two writers store the same deltas. For Xe/x, both keep 12,016 deltas and +both store 39 MiB. + +### Spill to disk by default? + +A run that sends every body to disk took 140 s, against 23 s. System time went +from 5 s to 109 s. Most objects are a few kilobytes, so each one costs file +system calls that are larger than the work. Each copy command in a delta also +becomes one `pread`. The peak heap fell by only 33 MiB, because the 64 MiB +budget already bounds the bodies in memory. + +## A bug that phase 1 found + +A stored delta chain of exactly 50 links did not read back. `deltaBase` +refused depth 50, but depth 50 is the link that reaches the whole object. Git +uses a default `pack.depth` of 50, so normal pushes make such chains. The +writer on `main` has the same fault. The memory problem hid it, because +golang/go never got far enough. + +A client can also send chains of up to 4095 links, and a read walks only 50. +Two changes correct this: + +- A read now accepts a chain of exactly `maxDeltaDepth` links. +- If a stored chain is already `maxDeltaDepth` links long, the writer stores + the next object whole. In a chain of 197 links, this stores 10 of 803 objects + whole. + +## Phase 2: make the pack index compact (next) + +After phase 1, the memory that remains grows with the number of objects, at +about 390 bytes for each live object. For golang/go that is about 275 MiB. +For a repository of 10 million objects, it is about 4 GB. + +| Part | Bytes per object | golang/go | +| ------------------------------ | ---------------: | --------: | +| `packIndex.entries` map | about 210 | 149 MiB | +| `cueRecord` slices of the push | about 70 | 48 MiB | +| push scan index | about 50 | 35 MiB | +| zstd history (fixed) | – | 64 MiB | + +The map is the largest part. Reads use it too: a clone or a fetch builds the +same map for every object in the repository. A smaller map therefore helps +both the push and the read path. + +### Design + +Replace `entries map[plumbing.Hash]packEntry` with a list of sorted tables. + +- A table holds the hashes of its objects as one `[]byte`, sorted, with + `hashSize` bytes for each hash. +- A parallel slice holds a small fixed-size record for each object: the pack + number, the offset, the stored length, the raw size, the type, and the + codec. +- The base of a delta is an index into the same table. A base in another + table goes in a small side map. The writer keeps a base and its delta in the + same container, so the side map is almost always empty. +- A table has a slice of pack ids. Each record holds a `uint32` index into + that slice, and not a `string`. + +This takes about 50 bytes for each object, against about 210 bytes now. + +`packLookup` does a binary search in each table. The cold build makes one +table from every `.cue`. Each container that a push registers adds one table. +When there are more than 16 tables, the index merges them into one table, so +a lookup never searches more than 16 tables. + +`deregister` removes a table that is still alone. For a table that a merge +absorbed, it adds the pack id to a set of dead packs, and a lookup skips a +record from a dead pack. `deregister` then needs only the pack id, so +`packJob` no longer holds its `[]cueRecord` until the upload completes. + +`snapshotEntries` keeps its order, by pack and then by offset. It reads that +order from the tables. + +`packEntry` stays as the type that `packLookup` returns. The lookup builds it +from a record, so callers do not change. + +### Order of work + +1. Add the table type with lookup, merge, and dead packs. Give it + table-driven tests. +2. Move `indexRecords`, `register`, `deregister`, `packLookup`, and + `snapshotEntries` to the tables. +3. Change `packJob` and `deregister` to use the pack id only. +4. Make the cold build parse each `.cue` into a table, and then release the + parsed records. +5. Measure the push and a cold index build of golang/go again. + +## Phase 3: work to examine later + +- **Index size for very large repositories.** After phase 2, about 110 bytes + for each object remain during a push. For 10 million objects that is about + 1.1 GB. A table can live in a memory-mapped temp file, and then the kernel + can page it out. +- **The zstd encoder.** It keeps 64 MiB of history for each process. + `EncodeAll` inputs are at most `inMemoryCap` (1 MiB), except `.cue` blocks. + A 1 MiB window cuts that to about 8 MiB. This changes the compressed bytes, + so it needs its own measurement. +- **Large bases in their own container.** An object larger than + `maxPackBytes` gets a container of its own. A delta against it lands in the + next container, so the writer stores that delta whole. This behavior is + older than this plan. diff --git a/docs/reference/tigris-backend.md b/docs/reference/tigris-backend.md index 15c64b8..8a414b3 100644 --- a/docs/reference/tigris-backend.md +++ b/docs/reference/tigris-backend.md @@ -95,23 +95,24 @@ the bytes against the name. ### The write path for packs -`PackfileWriter` writes the incoming pack to a scratch -`storage/filesystem.Storage` on a local temporary directory. That storer -decodes the pack. This package writes no pack-parsing code of its own. - -The writer makes two passes over the scratch objects. The first pass -reads the hash, the type, the size, and the delta base of each object. -It then puts the objects in an order where a base always comes before -the delta that needs it. This order is necessary because -`IterEncodedObjects` returns objects in hash order. - -The second pass copies each payload into the `.bin` and adds one record -to the `.cue`. The writer keeps the delta of the client when the base of -that delta is already in the container under construction. If the base -is not in that container, the writer stores the whole object instead. -This rule keeps every container complete on its own. - -The writer then deletes the scratch directory. The two files upload +`PackfileWriter` decodes the incoming pack itself, in `indexpack.go`. + +The first pass reads the pack as it arrives. It stages the bytes in a +temporary file and verifies the trailer checksum. It keeps a small index +entry for each object, and it keeps no body. + +The second pass walks each delta tree depth first, from its base. A base +therefore always comes before the deltas that need it. The pass copies +each payload into the `.bin` and adds one record to the `.cue`. The writer +keeps the delta of the client when the base of that delta is already in +the container under construction. If the base is not in that container, +the writer stores the whole object instead. This rule keeps every +container complete on its own. + +The writer also stores the whole object when the stored chain is already +`maxDeltaDepth` (50) links long. A read walks no more links than that. + +The writer then deletes its temporary files. The two files upload asynchronously. `SetReference` waits for those uploads, so a ref never points to a pack that the bucket does not hold. From b36f5655e794d413442fe2064afee2017f53ad37 Mon Sep 17 00:00:00 2001 From: Xe Iaso Date: Fri, 25 Sep 2026 16:47:33 -0400 Subject: [PATCH 3/8] perf(storage/tigris): replace the pack index map with sorted tables The pack index kept a map[plumbing.Hash]packEntry, about 210 bytes for each object. Every push held one entry for each object it wrote, and every clone or fetch built one for each object in the repository. Keep the entries in immutable tables instead: the hashes in one sorted []byte and a 40-byte record for each object, about 60 bytes with SHA-1. A record names its pack by number and its delta base by record number. Each registered container adds a table, and more than 16 tables merge into one, so a lookup is at most 16 binary searches. deregister takes only a pack id, so packJob no longer holds its cueRecords until the upload ends. The iterator walks 8-byte record references instead of a copy of every entry, and yields a hash only from the record the lookup returns, which removes its set of seen hashes. A snapshot shares the tables with the index rather than copying them, because the index never edits them in place. For golang/go, a cold index build drops from 225 MiB to 113 MiB of live heap (the index itself from about 180 MiB to 40 MiB). A push peaks at 364 MiB instead of 521 MiB. Assisted-by: Claude Opus 5.5 via Claude Code Signed-off-by: Xe Iaso --- internal/storage/tigris/iter.go | 35 +-- internal/storage/tigris/packindex.go | 255 +++++++++++++++++----- internal/storage/tigris/packtable.go | 216 ++++++++++++++++++ internal/storage/tigris/packtable_test.go | 220 +++++++++++++++++++ internal/storage/tigris/packwriter.go | 10 +- internal/storage/tigris/tigris.go | 4 +- 6 files changed, 659 insertions(+), 81 deletions(-) create mode 100644 internal/storage/tigris/packtable.go create mode 100644 internal/storage/tigris/packtable_test.go diff --git a/internal/storage/tigris/iter.go b/internal/storage/tigris/iter.go index 48f4728..673f19d 100644 --- a/internal/storage/tigris/iter.go +++ b/internal/storage/tigris/iter.go @@ -48,8 +48,8 @@ func (s *Storer) listKeys(prefix string) ([]string, error) { } // objectIter walks packed objects first — one whole pack at a time, in offset -// order, as snapshotEntries hands them over — then resolves loose keys one HEAD -// at a time. Laziness buys the cost profile the spec asks for: type mismatches +// order, as packSnapshot.order hands them over — then resolves loose keys one +// HEAD at a time. Laziness buys the cost profile the spec asks for: type mismatches // cost a HEAD, never a body download. // // Offset order is also what makes this cheap for a whole pack. The first packed @@ -60,40 +60,45 @@ func (s *Storer) listKeys(prefix string) ([]string, error) { type objectIter struct { s *Storer want plumbing.ObjectType - packed []packedEntry + // snap and packed replace a copy of every packEntry: packed costs 8 + // bytes for each object. A hash in more than one pack is yielded once, + // from the record that snap.lookup returns, and the loose walk skips any + // hash that snap holds, so no set of seen hashes is needed. + snap packSnapshot + packed []packRef ppos int keys []string pos int - seen map[plumbing.Hash]struct{} // packed hashes already yielded, so the loose walk can skip duplicates } func (s *Storer) IterEncodedObjects(t plumbing.ObjectType) (storer.EncodedObjectIter, error) { if err := s.ensurePacksBuilt(); err != nil { return nil, err } - packed := s.packs.snapshotEntries() + snap := s.packs.snapshot() keys, err := s.listKeys(s.prefix + objectPrefix) if err != nil { return nil, err } - return &objectIter{s: s, want: t, packed: packed, keys: keys, seen: make(map[plumbing.Hash]struct{}, len(packed))}, nil + return &objectIter{s: s, want: t, snap: snap, packed: snap.order(), keys: keys}, nil } func (it *objectIter) Next() (plumbing.EncodedObject, error) { for it.ppos < len(it.packed) { - pe := it.packed[it.ppos] + ref := it.packed[it.ppos] it.ppos++ - if _, dup := it.seen[pe.hash]; dup { - continue + t := it.snap.tables[ref.t] + hb := t.hashAt(int(ref.i)) + if ti, i, _ := it.snap.lookup(hb); ti != int(ref.t) || i != int(ref.i) { + continue // another pack holds the record that wins for this hash } - it.seen[pe.hash] = struct{}{} - - if it.want != plumbing.AnyObject && pe.e.typ != it.want { + if it.want != plumbing.AnyObject && t.recs[ref.i].typ != it.want { continue } - obj, err := it.s.packObject(it.want, pe.hash, pe.e) + h, _ := plumbing.FromBytes(hb) + obj, err := it.s.packObject(it.want, h, t.entry(int(ref.i), it.snap.ids)) if errors.Is(err, plumbing.ErrObjectNotFound) { continue // pack vanished between the index build and this read: tolerate, like the loose race below } @@ -111,8 +116,8 @@ func (it *objectIter) Next() (plumbing.EncodedObject, error) { if !ok { continue // junk under objects/: skip, never poison the walk } - if _, dup := it.seen[h]; dup { - continue // already yielded from a pack + if _, _, packed := it.snap.lookup(h.Bytes()); packed { + continue // a pack holds it, so the packed walk dealt with it } hs, herr := it.s.headInfo(h) diff --git a/internal/storage/tigris/packindex.go b/internal/storage/tigris/packindex.go index 729e2d0..912ee01 100644 --- a/internal/storage/tigris/packindex.go +++ b/internal/storage/tigris/packindex.go @@ -2,14 +2,16 @@ package tigris import ( "bytes" + "cmp" "encoding/binary" "errors" "fmt" "io" "log/slog" + "maps" "os" "runtime" - "sort" + "slices" "strings" "sync" "sync/atomic" @@ -360,22 +362,10 @@ type packEntry struct { base plumbing.Hash } -// indexRecords folds one pack's cue records into the entry map, returning the -// .bin's total length. Shared by register (a pack this instance just staged) -// and ensurePacksBuilt (a pack listed out of the bucket) so the two can never -// disagree about what an entry means. Callers hold p.mu. -func (p *packIndex) indexRecords(id string, recs []cueRecord) int64 { +// packExtent is the length of a pack's .bin: the end of its last payload. +func packExtent(recs []cueRecord) int64 { var size int64 for _, r := range recs { - p.entries[r.hash] = packEntry{ - id: id, - typ: r.typ, - codec: r.codec, - offset: r.offset, - stored: r.stored, - raw: r.raw, - base: r.base, - } if end := r.offset + r.stored; end > size { size = end } @@ -383,13 +373,6 @@ func (p *packIndex) indexRecords(id string, recs []cueRecord) int64 { return size } -// packedEntry pairs a hash with its packEntry, for iteration (a map alone -// loses the hash as a first-class value). -type packedEntry struct { - hash plumbing.Hash - e packEntry -} - // packStream is the in-progress view of a whole-pack download: a descriptor // over the file being filled, and how many of its bytes are committed. Reads // go through ReadAt, which is a pread and therefore safe alongside the writer @@ -433,21 +416,66 @@ type packAccess struct { // descendant's — gets its own packIndex, for the same isolation reasons as // its uploader: one repository's pack backlog, failures, or read history can // never affect another's. +// +// The entries live in packTables (packtable.go), about 60 bytes for each +// object. A lookup searches the tables from the newest to the oldest, so the +// newest record of a hash wins, and a push's own writes win over the cold +// build. type packIndex struct { - mu sync.Mutex - built bool - entries map[plumbing.Hash]packEntry - sizes map[string]int64 // packID -> total .bin length - local map[string]string // packID -> staged local .bin path while its upload is pending - access map[string]*packAccess + mu sync.Mutex + built bool + hashSize int + + ids []string // pack number -> pack id + idNum map[string]uint32 // pack id -> pack number + tables []*packTable // oldest first; see maxPackTables + // dead holds the packs that deregister removed from a table that also + // holds other packs. A lookup skips their records, and the next merge + // drops them. + dead map[uint32]struct{} + + sizes map[string]int64 // packID -> total .bin length + local map[string]string // packID -> staged local .bin path while its upload is pending + access map[string]*packAccess } -func newPackIndex() *packIndex { +// maxPackTables bounds how many tables a lookup searches. Each container a +// push registers adds a table. When there are more than this, addTable merges +// them all into one. +const maxPackTables = 16 + +func newPackIndex(hashSize int) *packIndex { return &packIndex{ - entries: make(map[plumbing.Hash]packEntry), - sizes: make(map[string]int64), - local: make(map[string]string), - access: make(map[string]*packAccess), + hashSize: hashSize, + idNum: make(map[string]uint32), + sizes: make(map[string]int64), + local: make(map[string]string), + access: make(map[string]*packAccess), + } +} + +// packNum returns the number of pack id, and gives it one if it has none. +// Callers hold p.mu. +func (p *packIndex) packNum(id string) uint32 { + if n, ok := p.idNum[id]; ok { + return n + } + n := uint32(len(p.ids)) + p.ids = append(p.ids, id) + p.idNum[id] = n + return n +} + +// addTable appends t as the newest table, and merges the tables when there +// are too many. Callers hold p.mu. +func (p *packIndex) addTable(t *packTable) { + if t.len() == 0 { + return + } + p.tables = append(p.tables, t) + if len(p.tables) > maxPackTables { + p.tables = []*packTable{mergeTables(p.hashSize, p.tables, p.dead)} + p.dead = nil } } @@ -456,10 +484,22 @@ func newPackIndex() *packIndex { // localPath). Call it before enqueueing the upload, mirroring writer.go's // "register pending before enqueue" rule: a read must never race ahead of // the write it depends on. +// +// The index keeps its own copy of what it needs, so the caller can drop +// recs after this returns. func (p *packIndex) register(id string, recs []cueRecord, localBin string) { p.mu.Lock() defer p.mu.Unlock() - p.sizes[id] = p.indexRecords(id, recs) + num := p.packNum(id) + if _, ok := p.dead[num]; ok { + // A pack that failed once and is written again is live. The map is + // shared with snapshots, so replace it. + dead := maps.Clone(p.dead) + delete(dead, num) + p.dead = dead + } + p.addTable(newPackTable(p.hashSize, num, recs)) + p.sizes[id] = packExtent(recs) p.local[id] = localBin } @@ -474,12 +514,31 @@ func (p *packIndex) markUploaded(id string) { // deregister removes a pack that failed to upload entirely: its entries must // not point at a pack S3 will never have. -func (p *packIndex) deregister(id string, recs []cueRecord) { +func (p *packIndex) deregister(id string) { p.mu.Lock() defer p.mu.Unlock() - for _, r := range recs { - if e, ok := p.entries[r.hash]; ok && e.id == id { - delete(p.entries, r.hash) + if num, ok := p.idNum[id]; ok { + // A new slice and a new map, never edits in place: a packSnapshot + // shares both. + kept := make([]*packTable, 0, len(p.tables)) + shared := false + for _, t := range p.tables { + switch { + case t.pack == int64(num): + continue // this table is only that pack + case t.pack < 0: + shared = true + } + kept = append(kept, t) + } + p.tables = kept + if shared { + dead := maps.Clone(p.dead) + if dead == nil { + dead = make(map[uint32]struct{}) + } + dead[num] = struct{}{} + p.dead = dead } } delete(p.sizes, id) @@ -499,25 +558,77 @@ func (p *packIndex) binSize(id string) int64 { return p.sizes[id] } -// snapshotEntries lists every indexed object, ordered by pack and then by -// offset within that pack. The order matters because a push above either -// write-side cap lands as several containers (see packwriter.go): map order -// would interleave them, so a full iteration would hold every container's bulk -// download open at once. Draining one pack at a time keeps that to one. -func (p *packIndex) snapshotEntries() []packedEntry { +// packSnapshot is the index as it was at one moment. The tables do not +// change, so it needs no lock. +// +// It shares its slices and map with the packIndex, so taking one costs no +// copy. That is safe because the packIndex never edits them in place: tables +// and ids only grow past the length a snapshot sees, and a change to the +// table list or the dead set replaces the slice or the map. +type packSnapshot struct { + tables []*packTable + ids []string + dead map[uint32]struct{} +} + +func (p *packIndex) snapshot() packSnapshot { p.mu.Lock() defer p.mu.Unlock() - out := make([]packedEntry, 0, len(p.entries)) - for h, e := range p.entries { - out = append(out, packedEntry{hash: h, e: e}) + return packSnapshot{tables: p.tables[:len(p.tables):len(p.tables)], ids: p.ids[:len(p.ids):len(p.ids)], dead: p.dead} +} + +// lookup finds the record of h: in the newest table that has it, skipping +// dead packs. +func (sn packSnapshot) lookup(h []byte) (ti, i int, ok bool) { + for ti := len(sn.tables) - 1; ti >= 0; ti-- { + t := sn.tables[ti] + i := t.find(h) + if i < 0 { + continue + } + if _, gone := sn.dead[t.recs[i].pack]; gone { + continue + } + return ti, i, true + } + return 0, 0, false +} + +// packRef names one record of a packSnapshot. +type packRef struct { + t uint32 + i int32 +} + +// order lists every live record, ordered by pack and then by offset within +// that pack. The order matters because a push above either write-side cap +// lands as several containers (see packwriter.go): hash order would interleave +// them, so a full iteration would hold every container's bulk download open at +// once. Draining one pack at a time keeps that to one. +// +// A hash that is in more than one pack appears more than once here. The +// iterator yields only the record that lookup returns. +func (sn packSnapshot) order() []packRef { + var n int + for _, t := range sn.tables { + n += t.len() } - sort.Slice(out, func(i, j int) bool { - if out[i].e.id != out[j].e.id { - return out[i].e.id < out[j].e.id + refs := make([]packRef, 0, n) + for ti, t := range sn.tables { + for i, r := range t.recs { + if _, gone := sn.dead[r.pack]; !gone { + refs = append(refs, packRef{uint32(ti), int32(i)}) + } + } + } + slices.SortFunc(refs, func(a, b packRef) int { + ra, rb := sn.tables[a.t].recs[a.i], sn.tables[b.t].recs[b.i] + if c := strings.Compare(sn.ids[ra.pack], sn.ids[rb.pack]); c != 0 { + return c } - return out[i].e.offset < out[j].e.offset + return cmp.Compare(ra.offset, rb.offset) }) - return out + return refs } func (p *packIndex) getAccess(id string) *packAccess { @@ -563,8 +674,13 @@ func (s *Storer) ensurePacksBuilt() error { return fmt.Errorf("tigris: list packs: %w", err) } + // Parse each .cue into a table of its own, and drop its records at once, + // so the cold build never holds a whole repository of cueRecords. hashLen := s.oh.Size() - byID := make(map[string][]cueRecord) + var ( + tables []*packTable + sizes = make(map[string]int64) + ) for _, k := range keys { if !strings.HasSuffix(k, cueSuffix) { continue @@ -579,7 +695,15 @@ func (s *Storer) ensurePacksBuilt() error { if err != nil { return fmt.Errorf("tigris: parse cue %s: %w", k, err) } - byID[id] = recs + s.packs.mu.Lock() + num := s.packs.packNum(id) + s.packs.mu.Unlock() + tables = append(tables, newPackTable(hashLen, num, recs)) + sizes[id] = packExtent(recs) + } + // One table for the whole repository makes each lookup one binary search. + if len(tables) > 1 { + tables = []*packTable{mergeTables(hashLen, tables, nil)} } s.packs.mu.Lock() @@ -587,8 +711,17 @@ func (s *Storer) ensurePacksBuilt() error { if s.packs.built { return nil // lost a race with a concurrent cold build on this instance } - for id, recs := range byID { - s.packs.sizes[id] = s.packs.indexRecords(id, recs) + // The cold tables go first, which makes them the oldest. The packs this + // instance registered itself are newer, so their records win. + s.packs.tables = append(tables, s.packs.tables...) + for id, n := range sizes { + if _, ok := s.packs.sizes[id]; !ok { + s.packs.sizes[id] = n + } + } + if len(s.packs.tables) > maxPackTables { + s.packs.tables = []*packTable{mergeTables(hashLen, s.packs.tables, s.packs.dead)} + s.packs.dead = nil } s.packs.built = true return nil @@ -600,10 +733,12 @@ func (s *Storer) packLookup(h plumbing.Hash) (packEntry, bool, error) { if err := s.ensurePacksBuilt(); err != nil { return packEntry{}, false, err } - s.packs.mu.Lock() - defer s.packs.mu.Unlock() - e, ok := s.packs.entries[h] - return e, ok, nil + sn := s.packs.snapshot() + ti, i, ok := sn.lookup(h.Bytes()) + if !ok { + return packEntry{}, false, nil + } + return sn.tables[ti].entry(i, sn.ids), true, nil } // startPackFetch kicks off the whole-container download of packs/.bin in diff --git a/internal/storage/tigris/packtable.go b/internal/storage/tigris/packtable.go new file mode 100644 index 0000000..0bf80f2 --- /dev/null +++ b/internal/storage/tigris/packtable.go @@ -0,0 +1,216 @@ +package tigris + +import ( + "bytes" + "slices" + "sort" + + "github.com/go-git/go-git/v6/plumbing" +) + +// A packTable is part of the packIndex: the records of one or more packs, +// sorted by hash. It replaces a map[plumbing.Hash]packEntry, which cost about +// 210 bytes for each object. A table costs hashSize bytes and one 40-byte +// tableRec for each object, which is about 60 bytes with SHA-1. +// +// A table never changes after newPackTable or mergeTables builds it. An +// iterator can therefore hold a table without a lock, and a merge builds a new +// table and does not edit the old ones. +type packTable struct { + hashSize int + hashes []byte // hashSize bytes for each record, sorted + recs []tableRec + + // farBase holds the base of a delta whose base is not in this table. The + // writer keeps a base and its delta in the same container, so this map is + // almost always empty. + farBase map[int32]plumbing.Hash + + // pack is the one pack number that every record has, or -1 when the + // table holds more than one pack. deregister can drop a table with one + // pack. It must mark the pack dead in a table with more. + pack int64 +} + +// tableRec is one object of a packTable. It is a packEntry without the two +// wide fields: the pack id is a number into packIndex.ids, and the base is a +// record number. +type tableRec struct { + offset int64 + stored int64 + raw int64 + base int32 // record number of the delta base, or baseNone, or baseFar + pack uint32 + typ plumbing.ObjectType + codec uint8 +} + +const ( + baseNone int32 = -1 // the payload is the whole object + baseFar int32 = -2 // the base is in farBase +) + +func (t *packTable) len() int { return len(t.recs) } + +func (t *packTable) hashAt(i int) []byte { + return t.hashes[i*t.hashSize : (i+1)*t.hashSize] +} + +func (t *packTable) hash(i int) plumbing.Hash { + h, _ := plumbing.FromBytes(t.hashAt(i)) + return h +} + +// find returns the record number of h, or -1. +func (t *packTable) find(h []byte) int { + i := sort.Search(len(t.recs), func(i int) bool { return bytes.Compare(t.hashAt(i), h) >= 0 }) + if i < len(t.recs) && bytes.Equal(t.hashAt(i), h) { + return i + } + return -1 +} + +// baseHash returns the base of record i, or the zero hash for a whole object. +func (t *packTable) baseHash(i int) plumbing.Hash { + switch b := t.recs[i].base; b { + case baseNone: + return plumbing.ZeroHash + case baseFar: + return t.farBase[int32(i)] + default: + return t.hash(int(b)) + } +} + +// entry builds the packEntry of record i. ids maps a pack number to its id. +func (t *packTable) entry(i int, ids []string) packEntry { + r := t.recs[i] + return packEntry{ + id: ids[r.pack], + typ: r.typ, + codec: r.codec, + offset: r.offset, + stored: r.stored, + raw: r.raw, + base: t.baseHash(i), + } +} + +// tableSource is one record for buildTable: its hash, its record with the +// base not yet numbered, and the hash of its base. +type tableSource struct { + hash []byte + rec tableRec + base plumbing.Hash +} + +// buildTable makes a table from n records. hashOf returns the hash of record +// i, and at returns all of it. The records can come in any order, and a hash +// can repeat. For a repeated hash, the last record wins, as it did when a +// later indexRecords call overwrote a map key. +func buildTable(hashSize, n int, hashOf func(i int) []byte, at func(i int) tableSource) *packTable { + order := make([]int32, n) + for i := range order { + order[i] = int32(i) + } + // A stable sort keeps repeated hashes in input order, so the dedupe below + // can keep the last one. + slices.SortStableFunc(order, func(a, b int32) int { + return bytes.Compare(hashOf(int(a)), hashOf(int(b))) + }) + + t := &packTable{ + hashSize: hashSize, + hashes: make([]byte, 0, n*hashSize), + recs: make([]tableRec, 0, n), + pack: -1, + } + src := make([]int32, 0, n) // input index of each kept record + for k, i := range order { + if k+1 < len(order) && bytes.Equal(hashOf(int(i)), hashOf(int(order[k+1]))) { + continue // a later record with the same hash wins + } + s := at(int(i)) + t.hashes = append(t.hashes, s.hash...) + t.recs = append(t.recs, s.rec) + src = append(src, i) + } + + for k := range t.recs { + b := at(int(src[k])).base + if b == plumbing.ZeroHash { + t.recs[k].base = baseNone + continue + } + if j := t.find(b.Bytes()); j >= 0 { + t.recs[k].base = int32(j) + continue + } + if t.farBase == nil { + t.farBase = make(map[int32]plumbing.Hash) + } + t.recs[k].base = baseFar + t.farBase[int32(k)] = b + } + + for k, r := range t.recs { + switch { + case k == 0: + t.pack = int64(r.pack) + case t.pack != int64(r.pack): + t.pack = -1 + } + } + return t +} + +// newPackTable makes a table from the cue records of one pack. +func newPackTable(hashSize int, pack uint32, recs []cueRecord) *packTable { + // ObjectID.Bytes copies, so copy every hash once rather than once for + // each comparison of the sort. + flat := make([]byte, 0, len(recs)*hashSize) + for _, r := range recs { + flat = append(flat, r.hash.Bytes()...) + } + hashOf := func(i int) []byte { return flat[i*hashSize : (i+1)*hashSize] } + return buildTable(hashSize, len(recs), hashOf, func(i int) tableSource { + r := recs[i] + return tableSource{ + hash: hashOf(i), + rec: tableRec{ + offset: r.offset, + stored: r.stored, + raw: r.raw, + pack: pack, + typ: r.typ, + codec: r.codec, + }, + base: r.base, + } + }) +} + +// mergeTables makes one table from tables, and drops every record whose pack +// is in dead. When a hash is in more than one table, the record from the +// later table wins, as it does in packIndex.lookup. +func mergeTables(hashSize int, tables []*packTable, dead map[uint32]struct{}) *packTable { + type ref struct { + t int32 + i int32 + } + var refs []ref + for ti, t := range tables { + for i, r := range t.recs { + if _, gone := dead[r.pack]; gone { + continue + } + refs = append(refs, ref{int32(ti), int32(i)}) + } + } + hashOf := func(k int) []byte { return tables[refs[k].t].hashAt(int(refs[k].i)) } + return buildTable(hashSize, len(refs), hashOf, func(k int) tableSource { + t := tables[refs[k].t] + i := int(refs[k].i) + return tableSource{hash: t.hashAt(i), rec: t.recs[i], base: t.baseHash(i)} + }) +} diff --git a/internal/storage/tigris/packtable_test.go b/internal/storage/tigris/packtable_test.go new file mode 100644 index 0000000..45ccfc3 --- /dev/null +++ b/internal/storage/tigris/packtable_test.go @@ -0,0 +1,220 @@ +package tigris + +import ( + "crypto/sha1" + "fmt" + "testing" + + "github.com/go-git/go-git/v6/plumbing" +) + +// th makes a distinct SHA-1 hash from a label. +func th(label string) plumbing.Hash { + sum := sha1.Sum([]byte(label)) + h, _ := plumbing.FromBytes(sum[:]) + return h +} + +// rec makes a cue record for th(label) at offset, as a delta against base +// when base is not empty. +func rec(label string, offset int64, base string) cueRecord { + r := cueRecord{hash: th(label), typ: plumbing.BlobObject, offset: offset, stored: 10, raw: 10} + if base != "" { + r.base = th(base) + } + return r +} + +// indexOp is one step against a packIndex. +type indexOp struct { + register string // pack id to register with recs + recs []cueRecord + deregister string // pack id to deregister +} + +// named registers n packs of one object each, called prefix00, prefix01, and +// so on. +func named(prefix string, n int) []indexOp { + var ops []indexOp + for i := range n { + id := fmt.Sprintf("%s%02d", prefix, i) + ops = append(ops, indexOp{register: id, recs: []cueRecord{rec(id, 0, "")}}) + } + return ops +} + +func TestPackIndexTables(t *testing.T) { + // many registers packs p00 onward, enough of them to force merges. + many := func(n int) []indexOp { return named("p", n) } + + type want struct { + label string + pack string // "" means the hash must be missing + offset int64 + base string + } + tests := []struct { + name string + ops []indexOp + want []want + wantTables int // 0 skips the check + wantOrder []string // labels in iteration order, after the dedupe + }{ + { + name: "one pack resolves bases inside the table", + ops: []indexOp{{register: "a", recs: []cueRecord{ + rec("x", 0, ""), rec("y", 10, "x"), rec("z", 20, "y"), + }}}, + want: []want{ + {label: "x", pack: "a", offset: 0}, + {label: "y", pack: "a", offset: 10, base: "x"}, + {label: "z", pack: "a", offset: 20, base: "y"}, + {label: "nope"}, + }, + wantTables: 1, + wantOrder: []string{"x", "y", "z"}, + }, + { + name: "a base outside the table goes to farBase", + ops: []indexOp{ + {register: "a", recs: []cueRecord{rec("x", 0, "")}}, + {register: "b", recs: []cueRecord{rec("y", 0, "x")}}, + }, + want: []want{ + {label: "x", pack: "a"}, + {label: "y", pack: "b", base: "x"}, + }, + wantOrder: []string{"x", "y"}, + }, + { + name: "the newest pack wins a repeated hash", + ops: []indexOp{ + {register: "a", recs: []cueRecord{rec("x", 0, ""), rec("only-a", 10, "")}}, + {register: "b", recs: []cueRecord{rec("x", 50, "")}}, + }, + want: []want{ + {label: "x", pack: "b", offset: 50}, + {label: "only-a", pack: "a", offset: 10}, + }, + // x is yielded once, from pack b. + wantOrder: []string{"only-a", "x"}, + }, + { + name: "deregister drops a table of one pack", + ops: []indexOp{ + {register: "a", recs: []cueRecord{rec("x", 0, "")}}, + {register: "b", recs: []cueRecord{rec("x", 50, ""), rec("y", 60, "")}}, + {deregister: "b"}, + }, + want: []want{ + {label: "x", pack: "a", offset: 0}, + {label: "y"}, + }, + wantTables: 1, + wantOrder: []string{"x"}, + }, + { + name: "more than maxPackTables merges into one table", + ops: many(maxPackTables + 1), + want: []want{ + {label: "p00", pack: "p00"}, + {label: fmt.Sprintf("p%02d", maxPackTables), pack: fmt.Sprintf("p%02d", maxPackTables)}, + }, + wantTables: 1, + }, + { + name: "deregister after a merge marks the pack dead", + ops: append(many(maxPackTables+1), indexOp{deregister: "p03"}), + want: []want{ + {label: "p03"}, + {label: "p04", pack: "p04"}, + }, + wantTables: 1, + }, + { + name: "a merge drops the records of a dead pack", + ops: append(append(many(maxPackTables+1), indexOp{deregister: "p03"}), + named("q", maxPackTables)...), + want: []want{ + {label: "p03"}, + {label: "p05", pack: "p05"}, + }, + wantTables: 1, + }, + { + name: "a pack written again after it failed is live", + ops: append(append(many(maxPackTables+1), indexOp{deregister: "p03"}), + indexOp{register: "p03", recs: []cueRecord{rec("p03", 7, "")}}), + want: []want{ + {label: "p03", pack: "p03", offset: 7}, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + p := newPackIndex(20) + for _, op := range tt.ops { + switch { + case op.register != "": + p.register(op.register, op.recs, "") + case op.deregister != "": + p.deregister(op.deregister) + } + } + + sn := p.snapshot() + if tt.wantTables != 0 && len(sn.tables) != tt.wantTables { + t.Errorf("got %d tables, want %d", len(sn.tables), tt.wantTables) + } + for _, w := range tt.want { + ti, i, ok := sn.lookup(th(w.label).Bytes()) + if w.pack == "" { + if ok { + t.Errorf("%s: found in %s, want missing", w.label, sn.tables[ti].entry(i, sn.ids).id) + } + continue + } + if !ok { + t.Errorf("%s: missing, want pack %s", w.label, w.pack) + continue + } + e := sn.tables[ti].entry(i, sn.ids) + if e.id != w.pack || e.offset != w.offset { + t.Errorf("%s: got pack %s offset %d, want pack %s offset %d", w.label, e.id, e.offset, w.pack, w.offset) + } + wantBase := plumbing.ZeroHash + if w.base != "" { + wantBase = th(w.base) + } + if e.base != wantBase { + t.Errorf("%s: got base %s, want %s", w.label, e.base, wantBase) + } + } + + if tt.wantOrder != nil { + labels := map[plumbing.Hash]string{} + for _, op := range tt.ops { + for _, r := range op.recs { + for _, l := range tt.wantOrder { + if th(l) == r.hash { + labels[r.hash] = l + } + } + } + } + var got []string + for _, ref := range sn.order() { + hb := sn.tables[ref.t].hashAt(int(ref.i)) + if ti, i, _ := sn.lookup(hb); ti != int(ref.t) || i != int(ref.i) { + continue + } + got = append(got, labels[sn.tables[ref.t].hash(int(ref.i))]) + } + if fmt.Sprint(got) != fmt.Sprint(tt.wantOrder) { + t.Errorf("got order %v, want %v", got, tt.wantOrder) + } + } + }) + } +} diff --git a/internal/storage/tigris/packwriter.go b/internal/storage/tigris/packwriter.go index 6ccd228..5746917 100644 --- a/internal/storage/tigris/packwriter.go +++ b/internal/storage/tigris/packwriter.go @@ -458,14 +458,13 @@ func (w *packWriter) seal(seg *packSegment) error { w.s.packs.register(id, seg.recs, seg.path) job := &packJob{ id: id, - recs: seg.recs, binPath: seg.path, cuePath: cuePath, binSize: seg.offset, cueSize: int64(len(cueBytes)), } if err := w.s.up.enqueue(w.s.ctx, job); err != nil { - w.s.packs.deregister(id, seg.recs) + w.s.packs.deregister(id) os.Remove(seg.path) os.Remove(cuePath) return err @@ -478,9 +477,12 @@ func (w *packWriter) seal(seg *packSegment) error { // every .cue they list) through the same uploader/bundler the loose-object // path uses (see upload.go), so one SetReference-triggered flush() covers // both kinds of upload and surfaces either kind of failure. +// +// It holds no cueRecords: register copied what reads need, and a failed +// upload deregisters by id alone. The records of a large push are therefore +// freed as soon as each container seals. type packJob struct { id string - recs []cueRecord binPath string cuePath string binSize int64 @@ -518,7 +520,7 @@ func (j *packJob) putFile(ctx context.Context, s *Storer, localPath, key string) func (j *packJob) done(s *Storer, err error) { if err != nil { - s.packs.deregister(j.id, j.recs) + s.packs.deregister(j.id) } else { s.packs.markUploaded(j.id) } diff --git a/internal/storage/tigris/tigris.go b/internal/storage/tigris/tigris.go index d39fe38..f657a10 100644 --- a/internal/storage/tigris/tigris.go +++ b/internal/storage/tigris/tigris.go @@ -291,7 +291,7 @@ func New(ctx context.Context, bucket string, opts ...Option) (*Storer, error) { s.client = c } s.up = newUploader(s) - s.packs = newPackIndex() + s.packs = newPackIndex(s.oh.Size()) s.refs = newRefCache() s.shallow = newShallowCache() s.fetchSem = make(chan struct{}, maxLivePackFetches) @@ -334,7 +334,7 @@ func (s *Storer) Scoped(prefix string) *Storer { cp.prefix = strings.TrimSuffix(cp.prefix, "/") + "/" + prefix + "/" } cp.up = newUploader(&cp) - cp.packs = newPackIndex() + cp.packs = newPackIndex(cp.oh.Size()) cp.refs = newRefCache() cp.shallow = newShallowCache() return &cp From 5c59e92049b95029a885250969166a8b5b54ac62 Mon Sep 17 00:00:00 2001 From: Xe Iaso Date: Fri, 25 Sep 2026 16:47:33 -0400 Subject: [PATCH 4/8] perf(storage/tigris): cut system calls on the push path A CPU profile of a golang/go push spent about 80% of its time in system calls. packSegment wrote each object to its file with no buffer, which cost 14.7 s of a 26 s push. Write through a 1 MiB buffer and count the bytes, so copyCompressed no longer seeks to measure a payload. The unused segmentStore reads from an open segment, so it flushes first. Each inflate also read its section through a 32 KiB buffer, so even a tiny object read 32 KiB. Read a section of 64 KiB or less whole, with one ReadAt of its own length. Wall time did not move measurably on a laptop: one pread per inflate now dominates. The plan records mmap of the staged pack as the next step. Assisted-by: Claude Opus 5.5 via Claude Code Signed-off-by: Xe Iaso --- internal/storage/tigris/indexpack.go | 41 ++++++++++++++++++--- internal/storage/tigris/packwriter.go | 49 ++++++++++++++++++------- internal/storage/tigris/segmentstore.go | 4 ++ 3 files changed, 75 insertions(+), 19 deletions(-) diff --git a/internal/storage/tigris/indexpack.go b/internal/storage/tigris/indexpack.go index a4aef40..3c8f910 100644 --- a/internal/storage/tigris/indexpack.go +++ b/internal/storage/tigris/indexpack.go @@ -406,25 +406,53 @@ func (p *incomingPack) open(i int32) (io.ReadCloser, error) { // object in a push is inflated at least twice and a fresh reader allocates a // 32 KiB window each time. type inflater struct { - br *bufio.Reader - zr io.ReadCloser + br *bufio.Reader + small []byte // a whole small section, read with one ReadAt + sr bytes.Reader // over small + zr io.ReadCloser // nil until first use } var inflaterPool sync.Pool +// smallSection is the largest compressed section that getInflater reads +// whole. Most objects are far smaller than this. Reading exactly the section +// costs one pread of its own length, where a 32 KiB buffer read 32 KiB for +// every object, however small. +const smallSection = 64 << 10 + // getInflater reads a zlib stream from r. A reader that is an io.ByteReader -// is used as is, so the stream is not read past its end; any other reader -// gets a buffer, which can read ahead. +// is used as is, so the stream is not read past its end. A small section is +// read whole. Any other reader gets a buffer, which can read ahead. func getInflater(r io.Reader) (*inflater, error) { - src := r v, _ := inflaterPool.Get().(*inflater) if v == nil { v = &inflater{br: bufio.NewReaderSize(nil, 32<<10)} } - if _, ok := r.(io.ByteReader); !ok { + + var src io.Reader + switch rr := r.(type) { + case io.ByteReader: + src = r + case *io.SectionReader: + if n := rr.Size(); n <= smallSection { + if int64(cap(v.small)) < n { + v.small = make([]byte, smallSection) + } + v.small = v.small[:n] + if _, err := rr.ReadAt(v.small, 0); err != nil && !errors.Is(err, io.EOF) { + return nil, err + } + v.sr.Reset(v.small) + src = &v.sr + break + } + v.br.Reset(r) + src = v.br + default: v.br.Reset(r) src = v.br } + if v.zr == nil { zr, err := zlib.NewReader(src) if err != nil { @@ -445,6 +473,7 @@ func (in *inflater) Read(p []byte) (int, error) { return in.zr.Read(p) } func (in *inflater) Close() error { err := in.zr.Close() in.br.Reset(nil) + in.sr.Reset(nil) inflaterPool.Put(in) return err } diff --git a/internal/storage/tigris/packwriter.go b/internal/storage/tigris/packwriter.go index 5746917..8243895 100644 --- a/internal/storage/tigris/packwriter.go +++ b/internal/storage/tigris/packwriter.go @@ -174,6 +174,13 @@ type packSegment struct { path string recs []cueRecord offset int64 + + // out buffers writes to file. Most objects are a few kilobytes, so an + // unbuffered file cost one write system call for each object, which was + // more than half of the time of a large push. written counts every byte + // given to out, so no Seek is needed to measure a payload. + out *bufio.Writer + written int64 } func newPackSegment(s *Storer) (*packSegment, error) { @@ -181,7 +188,21 @@ func newPackSegment(s *Storer) (*packSegment, error) { if err != nil { return nil, fmt.Errorf("create pack staging file: %w", err) } - return &packSegment{s: s, file: f, path: f.Name()}, nil + g := &packSegment{s: s, file: f, path: f.Name()} + g.out = bufio.NewWriterSize(segmentFile{g}, 1<<20) + return g, nil +} + +// segmentFile is the file under a packSegment's buffer. +type segmentFile struct{ g *packSegment } + +func (w segmentFile) Write(p []byte) (int, error) { return w.g.file.Write(p) } + +// Write adds p to the segment's .bin, through the buffer. +func (g *packSegment) Write(p []byte) (int, error) { + n, err := g.out.Write(p) + g.written += int64(n) + return n, err } // add appends one object to the segment's .bin — compressed or not, per @@ -254,7 +275,7 @@ func (g *packSegment) writePayload(obj plumbing.EncodedObject) (codec uint8, raw return 0, 0, 0, fmt.Errorf("read %s: %w", obj.Hash(), err) } body, compressed := compressBlock(plain) - n, err := g.file.Write(body) + n, err := g.Write(body) if err != nil { return 0, 0, 0, fmt.Errorf("write %s: %w", obj.Hash(), err) } @@ -317,7 +338,7 @@ func (g *packSegment) copyRaw(obj plumbing.EncodedObject, rd io.Reader) (int64, defer fresh.Close() rd = fresh } - n, err := io.Copy(g.file, rd) + n, err := io.Copy(g, rd) if err != nil { return n, fmt.Errorf("copy %s: %w", obj.Hash(), err) } @@ -341,13 +362,10 @@ func (g *packSegment) copyRawWithHead(obj plumbing.EncodedObject, head []byte, r // repository's history can still hold enough of them that the allocation // adds up. func (g *packSegment) copyCompressed(obj plumbing.EncodedObject, head []byte, rd io.Reader) (int64, int64, error) { - before, err := g.file.Seek(0, io.SeekCurrent) - if err != nil { - return 0, 0, fmt.Errorf("locate %s in pack staging file: %w", obj.Hash(), err) - } + before := g.written zw := streamEncPool.Get().(*zstd.Encoder) - zw.Reset(g.file) + zw.Reset(g) rawN, copyErr := io.Copy(zw, io.MultiReader(bytes.NewReader(head), rd)) closeErr := zw.Close() streamEncPool.Put(zw) @@ -358,16 +376,16 @@ func (g *packSegment) copyCompressed(obj plumbing.EncodedObject, head []byte, rd return 0, 0, fmt.Errorf("finish compressing %s: %w", obj.Hash(), closeErr) } - after, err := g.file.Seek(0, io.SeekCurrent) - if err != nil { - return 0, 0, fmt.Errorf("measure %s in pack staging file: %w", obj.Hash(), err) - } - return rawN, after - before, nil + return rawN, g.written - before, nil } // rewind discards whatever the current object wrote, returning the staging // file to the end of the last successfully added object. func (g *packSegment) rewind() error { + if err := g.out.Flush(); err != nil { + return fmt.Errorf("flush pack staging file: %w", err) + } + g.written = g.offset if err := g.file.Truncate(g.offset); err != nil { return fmt.Errorf("truncate pack staging file: %w", err) } @@ -419,6 +437,11 @@ func (g *packSegment) discard() { // the sibling .cue, and queues both for upload. It takes ownership of the // segment's staging files, so every error path here removes them itself. func (w *packWriter) seal(seg *packSegment) error { + if err := seg.out.Flush(); err != nil { + seg.file.Close() + os.Remove(seg.path) + return fmt.Errorf("tigris: flush pack staging file: %w", err) + } if err := seg.file.Close(); err != nil { os.Remove(seg.path) return fmt.Errorf("tigris: close pack staging file: %w", err) diff --git a/internal/storage/tigris/segmentstore.go b/internal/storage/tigris/segmentstore.go index f0c724e..93779ac 100644 --- a/internal/storage/tigris/segmentstore.go +++ b/internal/storage/tigris/segmentstore.go @@ -260,6 +260,10 @@ func (ss *segmentStore) EncodedObject(t plumbing.ObjectType, h plumbing.Hash) (p return nil, plumbing.ErrObjectNotFound } + // The segment buffers its writes, so the bytes can still be in the buffer. + if err := ref.seg.out.Flush(); err != nil { + return nil, fmt.Errorf("tigris: flush staging for %s: %w", h.String(), err) + } stored := make([]byte, ref.rec.stored) if _, err := ref.seg.file.ReadAt(stored, ref.rec.offset); err != nil { return nil, fmt.Errorf("tigris: read %s from staging: %w", h.String(), err) From 593dfdbfadcea4bf4247500e4d782c661f971385 Mon Sep 17 00:00:00 2001 From: Xe Iaso Date: Fri, 25 Sep 2026 16:47:33 -0400 Subject: [PATCH 5/8] docs: record the compact pack index and its measurements Describe the table-based pack index in the storer architecture, and add the phase 2 results and the remaining push-time work to the plan. Assisted-by: Claude Opus 5.5 via Claude Code Signed-off-by: Xe Iaso --- docs/architecture/tigris-storer.md | 35 ++++++++++++++++++++ docs/plans/push-memory.md | 51 ++++++++++++++++++++++++++++-- 2 files changed, 84 insertions(+), 2 deletions(-) diff --git a/docs/architecture/tigris-storer.md b/docs/architecture/tigris-storer.md index e8228c9..4bc951b 100644 --- a/docs/architecture/tigris-storer.md +++ b/docs/architecture/tigris-storer.md @@ -438,6 +438,41 @@ Reads check three tiers, in this order: The pack index is built once for each `Storer`, from the `packs/*.cue` files. Those files are small, and no `.bin` body is read to build it. +### The pack index (`packtable.go`) + +The index is a list of sorted tables, and not a map. A table holds the hashes +of its records in one `[]byte`, and a 40-byte `tableRec` for each record. That +is about 60 bytes for each object with SHA-1. The map it replaced took about +210 bytes. For golang/go, a cold build holds 40 MiB of tables, against about +180 MiB of map. + +A `tableRec` holds a pack number and not a pack id, and the base of a delta as +a record number and not a hash. A base in another table goes in the small +`farBase` map. The writer keeps a base and its delta in one container, so that +map is almost always empty. + +The cold build parses each `.cue` into a table and then drops its records. It +then merges all of the tables into one. Each container that a push registers +adds one table. When there are more than `maxPackTables` (16) tables, the index +merges them into one, so a lookup does at most 16 binary searches. A lookup +searches from the newest table to the oldest, so the newest record of a hash +wins. + +A table never changes after it is built. A `packSnapshot` therefore shares the +table list, the pack ids, and the dead set with the index, and costs no copy. +The index never edits these in place. It only appends past the length that a +snapshot sees, or it replaces the slice or the map. + +`deregister` takes only a pack id. It drops a table that holds only that pack. +In a merged table it adds the pack to the dead set, a lookup skips the record, +and the next merge drops it. `packJob` therefore keeps no `cueRecord`s while +its upload runs. + +`IterEncodedObjects` walks `packSnapshot.order`, which costs 8 bytes for each +object, in pack order and then offset order. It yields a hash only from the +record that the lookup returns, so it needs no set of hashes that it has +already yielded. + A packed read is one ranged `GetObject` over the record's `stored` span. There is no delta chain to reconstruct. When the record names the zstd codec, the read decompresses one frame; otherwise the bytes are already the object. diff --git a/docs/plans/push-memory.md b/docs/plans/push-memory.md index 1a70d0c..720b6f8 100644 --- a/docs/plans/push-memory.md +++ b/docs/plans/push-memory.md @@ -105,7 +105,7 @@ Two changes correct this: the next object whole. In a chain of 197 links, this stores 10 of 803 objects whole. -## Phase 2: make the pack index compact (next) +## Phase 2: make the pack index compact (done) After phase 1, the memory that remains grows with the number of objects, at about 390 bytes for each live object. For golang/go that is about 275 MiB. @@ -166,12 +166,59 @@ from a record, so callers do not change. parsed records. 5. Measure the push and a cold index build of golang/go again. +### Results of phase 2 + +The tables work as the design says, with two changes: + +- The iterator does not copy the entries. It keeps an 8-byte reference to each + record, and yields a hash only from the record that the lookup returns. This + removes the `seen` map, which cost more than 48 bytes for each object. +- A snapshot shares the table list, the ids, and the dead set with the index, + and does not copy them. A lookup takes a snapshot, so a copy there cost a + clone of every pack id on every lookup. + +| Measurement (golang/go) | Phase 1 | Phase 2 | +| -------------------------------- | ------: | ------: | +| Cold index build, live heap | 225 MiB | 113 MiB | +| Of which, the index itself | 180 MiB | 40 MiB | +| Push, peak heap | 521 MiB | 364 MiB | +| Push, live heap after the upload | 274 MiB | 114 MiB | + +The other 73 MiB of the cold build is fixed. 45 MiB is the buffers of the +`.cue` zstd decoder, and 28 MiB is the bodies that the S3 fake keeps. + +The round-trip test passes for all 708,000 objects of golang/go, and for every +other case of phase 1. The race detector finds no race in the package. + +### Two changes for push time + +A CPU profile of the push showed that system calls took about 80% of the +time. Two causes were in this package: + +- `packSegment` wrote to its file with no buffer. Each object cost at least one + `write` call, which was 14.7 s of a 26 s push. The segment now writes + through a 1 MiB buffer and counts its bytes, so `copyCompressed` needs no + `Seek`. +- Each inflate read its section through a 32 KiB buffer, so a small object + still read 32 KiB. A section of 64 KiB or less is now read with one `ReadAt` + of its own length. + +The wall time did not change by a measurable amount, at 25 s to 28 s on a +laptop. One `pread` for each inflate, about 1.4 million calls, now takes most +of the time. See phase 3. + ## Phase 3: work to examine later - **Index size for very large repositories.** After phase 2, about 110 bytes - for each object remain during a push. For 10 million objects that is about + for each object remain during a push: the scan index and the tables. For 10 million objects that is about 1.1 GB. A table can live in a memory-mapped temp file, and then the kernel can page it out. +- **One `pread` for each inflate.** A memory map of the staged pack removes + these calls. The mapped pages count in the RSS of the process, but they are + file pages that the kernel can drop. A `pread` puts the same pages in the + page cache, and cgroup accounting counts them in both cases. +- **The `.cue` zstd decoder.** It keeps about 45 MiB of buffers after its + first use. A lower decoder concurrency cuts that. - **The zstd encoder.** It keeps 64 MiB of history for each process. `EncodeAll` inputs are at most `inMemoryCap` (1 MiB), except `.cue` blocks. A 1 MiB window cuts that to about 8 MiB. This changes the compressed bytes, From b5a5f0b283e14d50f0efbd1bbe259ed71dac4bb2 Mon Sep 17 00:00:00 2001 From: Xe Iaso Date: Fri, 25 Sep 2026 16:55:02 -0400 Subject: [PATCH 6/8] style(storage/tigris): gofmt iter.go Assisted-by: Claude Opus 5.5 via Claude Code Signed-off-by: Xe Iaso --- internal/storage/tigris/iter.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/internal/storage/tigris/iter.go b/internal/storage/tigris/iter.go index 673f19d..b12422d 100644 --- a/internal/storage/tigris/iter.go +++ b/internal/storage/tigris/iter.go @@ -58,8 +58,8 @@ func (s *Storer) listKeys(prefix string) ([]string, error) { // same direction — so it catches up with the watermark early and reads the rest // locally, without ever having waited for the download. type objectIter struct { - s *Storer - want plumbing.ObjectType + s *Storer + want plumbing.ObjectType // snap and packed replace a copy of every packEntry: packed costs 8 // bytes for each object. A hash in more than one pack is yielded once, // from the record that snap.lookup returns, and the loose walk skips any From 52358c094b68b8d33b309f9b5960bd443c6597ac Mon Sep 17 00:00:00 2001 From: Xe Iaso Date: Fri, 25 Sep 2026 16:55:02 -0400 Subject: [PATCH 7/8] test(storage/tigris): cover the pack reader with hand-built packs The existing pack tests reach the new pack reader only through Write, with packs from the git CLI, so they never send a REF-delta, a delta whose base is already in the repository, a chain deeper than a read walks, or a malformed pack. Build packs by hand and push each one through both ReadPack and Write. The round-trip cases read every object back through a new Storer and check the stored delta count and the longest stored chain. The reject cases check the error and that no PutObject call was made. One more test pins that ReadPack leaves the bytes after the trailer unread, which writePack depends on for git:// and SSH. Removing the chain-depth cap or the delta source-size check makes these tests fail. Assisted-by: Claude Opus 5.5 via Claude Code Signed-off-by: Xe Iaso --- internal/storage/tigris/indexpack_test.go | 530 ++++++++++++++++++++++ 1 file changed, 530 insertions(+) create mode 100644 internal/storage/tigris/indexpack_test.go diff --git a/internal/storage/tigris/indexpack_test.go b/internal/storage/tigris/indexpack_test.go new file mode 100644 index 0000000..ccaa6ad --- /dev/null +++ b/internal/storage/tigris/indexpack_test.go @@ -0,0 +1,530 @@ +package tigris + +import ( + "bufio" + "bytes" + "compress/zlib" + "crypto/sha1" + "encoding/binary" + "errors" + "fmt" + "io" + "strings" + "testing" + + "github.com/go-git/go-git/v6/plumbing" + formatcfg "github.com/go-git/go-git/v6/plumbing/format/config" + "github.com/go-git/go-git/v6/plumbing/format/packfile" +) + +// entrySpec is one entry of a pack that buildTestPack writes by hand. Real +// git never sends some of these shapes, such as a REF-delta whose base comes +// later in the pack, so the tests cannot get them from the git CLI. +type entrySpec struct { + typ plumbing.ObjectType // the resolved type; blob when zero + body []byte // the resolved object + + ofs int // for an OFS-delta: 1 + the index of the base entry + ref plumbing.Hash // for a REF-delta: the base hash + // base is the resolved body of the base, for a delta. The delta itself + // copies all of base and then inserts the rest of body. + base []byte + + // rawDelta, when set, is written as the delta instead of one built from + // base and body. + rawDelta []byte + // declared, when not zero, replaces the inflated size in the header. + declared int64 + // ofsDist, when not zero, replaces the OFS-delta distance. + ofsDist int64 +} + +func (e entrySpec) objType() plumbing.ObjectType { + if e.typ == 0 { + return plumbing.BlobObject + } + return e.typ +} + +func (e entrySpec) hash() plumbing.Hash { + hs := plumbing.NewHasher(formatcfg.DefaultObjectFormat, e.objType(), int64(len(e.body))) + hs.Write(e.body) + return hs.Sum() +} + +// testDelta builds a git delta that copies all of base and then inserts the +// rest of target. target must start with base. +func testDelta(t *testing.T, base, target []byte) []byte { + t.Helper() + if !bytes.HasPrefix(target, base) { + t.Fatalf("testDelta: target does not start with base") + } + var d []byte + putSize := func(n int) { + for { + b := byte(n & 0x7f) + n >>= 7 + if n == 0 { + d = append(d, b) + return + } + d = append(d, b|0x80) + } + } + putSize(len(base)) + putSize(len(target)) + + // Copy commands, at most 0xffffff bytes each. + for off := 0; off < len(base); { + n := min(len(base)-off, 0xffffff) + cmd := byte(0x80) + var args []byte + for k := range 4 { + if b := byte(off >> (8 * k)); b != 0 { + cmd |= 1 << k + args = append(args, b) + } + } + for k := range 3 { + if b := byte(n >> (8 * k)); b != 0 { + cmd |= 0x10 << k + args = append(args, b) + } + } + d = append(d, cmd) + d = append(d, args...) + off += n + } + // Insert commands, at most 127 bytes each. + rest := target[len(base):] + for len(rest) > 0 { + n := min(len(rest), 127) + d = append(d, byte(n)) + d = append(d, rest[:n]...) + rest = rest[n:] + } + return d +} + +func deflate(t *testing.T, b []byte) []byte { + t.Helper() + var buf bytes.Buffer + zw := zlib.NewWriter(&buf) + if _, err := zw.Write(b); err != nil { + t.Fatal(err) + } + if err := zw.Close(); err != nil { + t.Fatal(err) + } + return buf.Bytes() +} + +// buildTestPack writes a version 2 packfile of entries, with a correct +// trailer. +func buildTestPack(t *testing.T, entries []entrySpec) []byte { + t.Helper() + var pack bytes.Buffer + pack.WriteString("PACK") + binary.Write(&pack, binary.BigEndian, uint32(2)) + binary.Write(&pack, binary.BigEndian, uint32(len(entries))) + + offsets := make([]int64, len(entries)) + for i, e := range entries { + offsets[i] = int64(pack.Len()) + + payload := e.body + typ := e.objType() + switch { + case e.ofs != 0: + typ = plumbing.OFSDeltaObject + case e.ref != plumbing.ZeroHash: + typ = plumbing.REFDeltaObject + } + if typ.IsDelta() { + payload = e.rawDelta + if payload == nil { + payload = testDelta(t, e.base, e.body) + } + } + + size := int64(len(payload)) + if e.declared != 0 { + size = e.declared + } + c := byte(typ)<<4 | byte(size&0x0f) + size >>= 4 + for size != 0 { + pack.WriteByte(c | 0x80) + c = byte(size & 0x7f) + size >>= 7 + } + pack.WriteByte(c) + + switch typ { + case plumbing.OFSDeltaObject: + dist := offsets[i] - offsets[e.ofs-1] + if e.ofsDist != 0 { + dist = e.ofsDist + } + var enc [10]byte + pos := len(enc) - 1 + enc[pos] = byte(dist & 0x7f) + for dist >>= 7; dist != 0; dist >>= 7 { + dist-- + pos-- + enc[pos] = 0x80 | byte(dist&0x7f) + } + pack.Write(enc[pos:]) + case plumbing.REFDeltaObject: + pack.Write(e.ref.Bytes()) + } + pack.Write(deflate(t, payload)) + } + sum := sha1.Sum(pack.Bytes()) + pack.Write(sum[:]) + return pack.Bytes() +} + +// pushPack stores pack through s, the way writePack in cmd/objgitd does +// when viaReadPack is true, and through Write when it is false. +func pushPack(t *testing.T, s *Storer, pack []byte, viaReadPack bool) error { + t.Helper() + w, err := s.PackfileWriter() + if err != nil { + t.Fatalf("PackfileWriter: %v", err) + } + if viaReadPack { + if err := w.(*packWriter).ReadPack(bufio.NewReader(bytes.NewReader(pack))); err != nil { + _ = w.Close() + return err + } + } else if _, err := w.Write(pack); err != nil { + t.Fatalf("Write: %v", err) + } + if err := w.Close(); err != nil { + return err + } + return s.up.flush() +} + +// verifyReadBack opens a new Storer over f and verifies that every entry reads +// back whole, with the right type and body. +func verifyReadBack(t *testing.T, f *fakeS3, prefix string, entries []entrySpec) *Storer { + t.Helper() + cold := newTestStorer(t, f).Scoped(prefix) + for _, e := range entries { + h := e.hash() + obj, err := cold.EncodedObject(plumbing.AnyObject, h) + if err != nil { + t.Fatalf("EncodedObject(%s): %v", h, err) + } + if obj.Type() != e.objType() { + t.Errorf("%s: type %s, want %s", h, obj.Type(), e.objType()) + } + rd, err := obj.Reader() + if err != nil { + t.Fatalf("%s: Reader: %v", h, err) + } + got, err := io.ReadAll(rd) + rd.Close() + if err != nil { + t.Fatalf("%s: read: %v", h, err) + } + if !bytes.Equal(got, e.body) { + t.Errorf("%s: body of %d bytes differs from the %d bytes pushed", h, len(got), len(e.body)) + } + } + return cold +} + +// storedChains reports how many records are stored as deltas, and the +// longest stored delta chain, over every table of s. +func storedChains(t *testing.T, s *Storer) (deltas, longest int) { + t.Helper() + if err := s.ensurePacksBuilt(); err != nil { + t.Fatal(err) + } + sn := s.packs.snapshot() + for _, tb := range sn.tables { + for i := range tb.recs { + n := 0 + for h := tb.baseHash(i); h != plumbing.ZeroHash; n++ { + ti, j, ok := sn.lookup(h.Bytes()) + if !ok { + t.Fatalf("base %s is not indexed", h) + } + h = sn.tables[ti].baseHash(j) + } + if n > 0 { + deltas++ + } + longest = max(longest, n) + } + } + return deltas, longest +} + +// chain returns n+1 entries: a whole blob, then n OFS-deltas, each one on +// the entry before it. +func chain(n int) []entrySpec { + body := []byte("line 0\n") + out := []entrySpec{{body: body}} + for i := 1; i <= n; i++ { + next := append(bytes.Clone(body), fmt.Sprintf("line %d\n", i)...) + out = append(out, entrySpec{body: next, base: body, ofs: i}) + body = next + } + return out +} + +func TestIndexPackRoundTrip(t *testing.T) { + a := []byte(strings.Repeat("a base blob\n", 20)) + ab := append(bytes.Clone(a), "and a line on top\n"...) + abc := append(bytes.Clone(ab), "and one more\n"...) + big := bytes.Repeat([]byte("0123456789abcdef"), 1<<14) // 256 KiB + bigger := append(bytes.Clone(big), "tail"...) + + blobA := entrySpec{body: a} + + tests := []struct { + name string + existing []entrySpec // pushed first, in a pack of its own + entries []entrySpec + spill bool // send every body to disk + wantDeltas int // stored delta records; -1 skips the check + wantMaxLen int // the longest stored chain may not be longer + }{ + { + name: "whole objects of every type", + entries: []entrySpec{ + {body: []byte("a blob\n")}, + {typ: plumbing.TreeObject, body: []byte("not a real tree")}, + {typ: plumbing.CommitObject, body: []byte("not a real commit")}, + {typ: plumbing.TagObject, body: []byte("not a real tag")}, + {body: []byte{}}, + }, + wantDeltas: 0, + }, + { + name: "ofs-delta chain", + entries: chain(5), + wantDeltas: 5, + wantMaxLen: 5, + }, + { + name: "ref-delta after its base", + entries: []entrySpec{ + blobA, + {body: ab, base: a, ref: blobA.hash()}, + }, + wantDeltas: 1, + wantMaxLen: 1, + }, + { + name: "ref-delta before its base", + entries: []entrySpec{ + {body: ab, base: a, ref: blobA.hash()}, + blobA, + }, + wantDeltas: 1, + wantMaxLen: 1, + }, + { + name: "ref-delta on a ref-delta", + entries: []entrySpec{ + {body: abc, base: ab, ref: entrySpec{body: ab}.hash()}, + {body: ab, base: a, ref: blobA.hash()}, + blobA, + }, + wantDeltas: 2, + wantMaxLen: 2, + }, + { + name: "ref-delta on a base already in the repository", + existing: []entrySpec{blobA}, + entries: []entrySpec{ + {body: ab, base: a, ref: blobA.hash()}, + }, + // The base is in another container, so the delta is stored whole. + wantDeltas: 0, + }, + { + name: "a chain deeper than a read walks", + entries: chain(3*maxDeltaDepth + 7), + wantDeltas: -1, + wantMaxLen: maxDeltaDepth, + }, + { + name: "a chain of exactly maxDeltaDepth links", + entries: chain(maxDeltaDepth), + wantDeltas: maxDeltaDepth, + wantMaxLen: maxDeltaDepth, + }, + { + name: "every body on disk", + entries: append(chain(12), + entrySpec{body: big}, + entrySpec{body: bigger, base: big, ofs: 14}, + ), + spill: true, + wantDeltas: 13, + wantMaxLen: 12, + }, + } + + for _, tt := range tests { + for _, viaReadPack := range []bool{true, false} { + name := tt.name + "/write" + if viaReadPack { + name = tt.name + "/readpack" + } + t.Run(name, func(t *testing.T) { + if tt.spill { + limit, budget := bodyMemLimit, bodyMemBudget + bodyMemLimit, bodyMemBudget = 0, 0 + t.Cleanup(func() { bodyMemLimit, bodyMemBudget = limit, budget }) + } + + f := newFakeS3(t) + s := newTestStorer(t, f).Scoped("org/repo") + if tt.existing != nil { + if err := pushPack(t, s, buildTestPack(t, tt.existing), viaReadPack); err != nil { + t.Fatalf("push existing: %v", err) + } + } + if err := pushPack(t, s, buildTestPack(t, tt.entries), viaReadPack); err != nil { + t.Fatalf("push: %v", err) + } + + cold := verifyReadBack(t, f, "org/repo", append(tt.existing, tt.entries...)) + deltas, longest := storedChains(t, cold) + if tt.wantDeltas >= 0 && deltas != tt.wantDeltas { + t.Errorf("got %d stored deltas, want %d", deltas, tt.wantDeltas) + } + if longest > tt.wantMaxLen { + t.Errorf("longest stored chain is %d links, want at most %d", longest, tt.wantMaxLen) + } + }) + } + } +} + +func TestIndexPackRejects(t *testing.T) { + a := []byte("a base blob\n") + ab := append(bytes.Clone(a), "on top\n"...) + good := func(t *testing.T) []byte { + return buildTestPack(t, []entrySpec{{body: a}, {body: ab, base: a, ofs: 1}}) + } + + tests := []struct { + name string + pack func(t *testing.T) []byte + wantErr error // nil only checks that there is an error + }{ + { + name: "bad signature", + pack: func(t *testing.T) []byte { + p := good(t) + copy(p, "KCAP") + return p + }, + wantErr: packfile.ErrMalformedPackfile, + }, + { + name: "bad trailer checksum", + pack: func(t *testing.T) []byte { + p := good(t) + p[len(p)-1] ^= 0xff + return p + }, + wantErr: packfile.ErrMalformedPackfile, + }, + { + name: "truncated", + pack: func(t *testing.T) []byte { + p := good(t) + return p[:len(p)-30] + }, + }, + { + name: "ofs-delta names no entry", + pack: func(t *testing.T) []byte { + return buildTestPack(t, []entrySpec{{body: a}, {body: ab, base: a, ofs: 1, ofsDist: 3}}) + }, + wantErr: packfile.ErrMalformedPackfile, + }, + { + name: "entry shorter than its header says", + pack: func(t *testing.T) []byte { + return buildTestPack(t, []entrySpec{{body: a, declared: int64(len(a)) + 5}}) + }, + wantErr: packfile.ErrMalformedPackfile, + }, + { + name: "delta names the wrong base size", + pack: func(t *testing.T) []byte { + d := testDelta(t, a, ab) + d[0]++ // the source size + return buildTestPack(t, []entrySpec{{body: a}, {body: ab, ofs: 1, rawDelta: d}}) + }, + wantErr: packfile.ErrInvalidDelta, + }, + { + name: "ref-delta base missing", + pack: func(t *testing.T) []byte { + return buildTestPack(t, []entrySpec{{body: ab, base: a, ref: entrySpec{body: a}.hash()}}) + }, + wantErr: plumbing.ErrObjectNotFound, + }, + } + + for _, tt := range tests { + for _, viaReadPack := range []bool{true, false} { + name := tt.name + "/write" + if viaReadPack { + name = tt.name + "/readpack" + } + t.Run(name, func(t *testing.T) { + f := newFakeS3(t) + s := newTestStorer(t, f).Scoped("org/repo") + err := pushPack(t, s, tt.pack(t), viaReadPack) + if err == nil { + t.Fatal("push succeeded, want an error") + } + if tt.wantErr != nil && !errors.Is(err, tt.wantErr) { + t.Errorf("error = %v, want %v", err, tt.wantErr) + } + if n := f.nputs(); n != 0 { + t.Errorf("a rejected push made %d PutObject calls, want 0", n) + } + }) + } + } +} + +// TestReadPackStopsAtTrailer pins the property that writePack depends on for +// git:// and SSH: ReadPack consumes the pack and not one byte more. +func TestReadPackStopsAtTrailer(t *testing.T) { + pack := buildTestPack(t, chain(3)) + after := []byte("0000 bytes the client sends after the pack") + + s := newTestStorer(t, newFakeS3(t)).Scoped("org/repo") + w, err := s.PackfileWriter() + if err != nil { + t.Fatal(err) + } + br := bufio.NewReader(bytes.NewReader(append(bytes.Clone(pack), after...))) + if err := w.(*packWriter).ReadPack(br); err != nil { + t.Fatalf("ReadPack: %v", err) + } + if err := w.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + rest, err := io.ReadAll(br) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(rest, after) { + t.Errorf("left %q unread, want %q", rest, after) + } +} From e9979b496304f5698061a9c190edf4e41d938826 Mon Sep 17 00:00:00 2001 From: Xe Iaso Date: Fri, 25 Sep 2026 16:55:41 -0400 Subject: [PATCH 8/8] docs: format the push memory plan with prettier Assisted-by: Claude Opus 5.5 via Claude Code Signed-off-by: Xe Iaso --- docs/plans/push-memory.md | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/docs/plans/push-memory.md b/docs/plans/push-memory.md index 720b6f8..a9c2796 100644 --- a/docs/plans/push-memory.md +++ b/docs/plans/push-memory.md @@ -65,13 +65,13 @@ where the client keeps the connection open after the pack. The numbers come from a local harness. It feeds a real pack to `PackfileWriter` over an S3 fake that discards bodies. -| Push | Before | After | -| ---------------------------- | ------------------------ | ------------------------- | -| golang/go, peak heap | more than 21 GiB | 521 MiB | -| golang/go, time | did not complete | 23 s | -| Xe/x, peak heap | 516 MiB | about 165 MiB | -| Xe/x, time | 14.7 s | 1.3 s | -| 266 MB blob and one delta | both bodies in memory | 35 MiB peak heap | +| Push | Before | After | +| ------------------------- | --------------------- | ---------------- | +| golang/go, peak heap | more than 21 GiB | 521 MiB | +| golang/go, time | did not complete | 23 s | +| Xe/x, peak heap | 516 MiB | about 165 MiB | +| Xe/x, time | 14.7 s | 1.3 s | +| 266 MB blob and one delta | both bodies in memory | 35 MiB peak heap | A second test reads every object back through a new `Storer` and verifies its hash. It passes for all 708,000 objects of golang/go. It also passes for a @@ -113,7 +113,7 @@ For a repository of 10 million objects, it is about 4 GB. | Part | Bytes per object | golang/go | | ------------------------------ | ---------------: | --------: | -| `packIndex.entries` map | about 210 | 149 MiB | +| `packIndex.entries` map | about 210 | 149 MiB | | `cueRecord` slices of the push | about 70 | 48 MiB | | push scan index | about 50 | 35 MiB | | zstd history (fixed) | – | 64 MiB |