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/docs/architecture/tigris-storer.md b/docs/architecture/tigris-storer.md index 5906384..4bc951b 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 @@ -417,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 new file mode 100644 index 0000000..a9c2796 --- /dev/null +++ b/docs/plans/push-memory.md @@ -0,0 +1,229 @@ +# 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 (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. +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. + +### 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: 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, + 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. 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..3c8f910 --- /dev/null +++ b/internal/storage/tigris/indexpack.go @@ -0,0 +1,940 @@ +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 + 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. A small section is +// read whole. Any other reader gets a buffer, which can read ahead. +func getInflater(r io.Reader) (*inflater, error) { + v, _ := inflaterPool.Get().(*inflater) + if v == nil { + v = &inflater{br: bufio.NewReaderSize(nil, 32<<10)} + } + + 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 { + 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) + in.sr.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/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) + } +} diff --git a/internal/storage/tigris/iter.go b/internal/storage/tigris/iter.go index 48f4728..b12422d 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 @@ -58,42 +58,47 @@ 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 - packed []packedEntry + 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 + // 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 d88b5da..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 @@ -848,8 +983,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 +1044,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/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 63e6f28..8243895 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) { + n, err := w.buf.Write(p) + w.n += int64(n) + return n, err } -func (w *packWriter) Write(p []byte) (int, error) { return w.inner.Write(p) } +// 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. // @@ -397,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) { @@ -404,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 @@ -477,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) } @@ -540,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) } @@ -564,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) @@ -581,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) } @@ -642,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) @@ -681,14 +481,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 @@ -701,9 +500,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 @@ -741,7 +543,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/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) 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