aboutsummaryrefslogtreecommitdiff
path: root/object
diff options
context:
space:
mode:
Diffstat (limited to 'object')
-rw-r--r--object/parse.go4
-rw-r--r--object/store/dual/quarantine.go11
-rw-r--r--object/store/dual/writer.go5
-rw-r--r--object/store/loose/quarantine.go2
-rw-r--r--object/store/packed/internal/ingest/basecache.go25
-rw-r--r--object/store/packed/internal/ingest/finalize.go69
-rw-r--r--object/store/packed/internal/ingest/ingest.go57
-rw-r--r--object/store/packed/internal/ingest/record.go6
-rw-r--r--object/store/packed/internal/ingest/resolve.go339
-rw-r--r--object/store/packed/internal/ingest/scan.go24
-rw-r--r--object/store/packed/internal/ingest/thin.go10
-rw-r--r--object/store/packed/internal/ingest/writepack_test.go123
-rw-r--r--object/store/packed/quarantine.go35
-rw-r--r--object/store/packed/writer.go5
-rw-r--r--object/store/packed/writer_test.go2
-rw-r--r--object/store/writer.go14
-rw-r--r--object/stored/stored.go2
17 files changed, 569 insertions, 164 deletions
diff --git a/object/parse.go b/object/parse.go
index f9779171..2a6c629f 100644
--- a/object/parse.go
+++ b/object/parse.go
@@ -20,8 +20,6 @@ var ErrSizeMismatch = errors.New("object: size mismatch")
// ParseWithHeader parses a loose object
// in "type size\x00body" format.
-//
-//nolint:ireturn
func ParseWithHeader(raw []byte, objectFormat id.ObjectFormat) (Object, error) {
ty, size, headerLen, err := header.Parse(raw)
if err != nil {
@@ -37,8 +35,6 @@ func ParseWithHeader(raw []byte, objectFormat id.ObjectFormat) (Object, error) {
}
// ParseWithoutHeader parses a typed object body.
-//
-//nolint:ireturn
func ParseWithoutHeader(ty typ.Type, body []byte, objectFormat id.ObjectFormat) (Object, error) {
switch ty {
case typ.Blob:
diff --git a/object/store/dual/quarantine.go b/object/store/dual/quarantine.go
index b73e48fe..6052c134 100644
--- a/object/store/dual/quarantine.go
+++ b/object/store/dual/quarantine.go
@@ -1,6 +1,7 @@
package dual
import (
+ "context"
"errors"
"fmt"
"io"
@@ -12,15 +13,11 @@ import (
)
// BeginObjectQuarantine begins an object-wise quarantine on the object side.
-//
-//nolint:ireturn
func (dual *Dual) BeginObjectQuarantine(opts store.ObjectQuarantineOptions) (store.ObjectQuarantine, error) {
return dual.object.BeginObjectQuarantine(opts) //nolint:wrapcheck
}
// BeginPackQuarantine begins a pack-wise quarantine on the pack side.
-//
-//nolint:ireturn
func (dual *Dual) BeginPackQuarantine(opts store.PackQuarantineOptions) (store.PackQuarantine, error) {
return dual.pack.BeginPackQuarantine(opts) //nolint:wrapcheck
}
@@ -29,8 +26,6 @@ func (dual *Dual) BeginPackQuarantine(opts store.PackQuarantineOptions) (store.P
//
// If the pack side fails to begin,
// the already-begun object side is discarded before returning.
-//
-//nolint:ireturn
func (dual *Dual) BeginCoordinatedQuarantine(opts store.CoordinatedQuarantineOptions) (store.CoordinatedQuarantine, error) {
objectQ, err := dual.object.BeginObjectQuarantine(opts.Object)
if err != nil {
@@ -111,8 +106,8 @@ func (quarantine *coordinatedQuarantine) WriteReaderContent(ty typ.Type, size in
return quarantine.objectQ.WriteReaderContent(ty, size, src) //nolint:wrapcheck
}
-func (quarantine *coordinatedQuarantine) WritePack(src io.Reader, opts store.PackWriteOptions) error {
- return quarantine.packQ.WritePack(src, opts) //nolint:wrapcheck
+func (quarantine *coordinatedQuarantine) WritePack(ctx context.Context, src io.Reader, opts store.PackWriteOptions) error {
+ return quarantine.packQ.WritePack(ctx, src, opts) //nolint:wrapcheck
}
// Promote publishes both halves and joins their errors.
diff --git a/object/store/dual/writer.go b/object/store/dual/writer.go
index f75f49e1..fb59adbe 100644
--- a/object/store/dual/writer.go
+++ b/object/store/dual/writer.go
@@ -1,6 +1,7 @@
package dual
import (
+ "context"
"io"
"lindenii.org/go/furgit/object/id"
@@ -29,6 +30,6 @@ func (dual *Dual) WriteReaderContent(ty typ.Type, size int, src io.Reader) (id.O
}
// WritePack ingests one pack stream into the pack side.
-func (dual *Dual) WritePack(src io.Reader, opts store.PackWriteOptions) error {
- return dual.pack.WritePack(src, opts) //nolint:wrapcheck
+func (dual *Dual) WritePack(ctx context.Context, src io.Reader, opts store.PackWriteOptions) error {
+ return dual.pack.WritePack(ctx, src, opts) //nolint:wrapcheck
}
diff --git a/object/store/loose/quarantine.go b/object/store/loose/quarantine.go
index 214f7219..cd337670 100644
--- a/object/store/loose/quarantine.go
+++ b/object/store/loose/quarantine.go
@@ -30,7 +30,7 @@ type objectQuarantine struct {
// beneath the destination loose root.
//
// Labels: Deps-Borrowed, Life-Parent, Close-No.
-func (loose *Loose) BeginObjectQuarantine(_ store.ObjectQuarantineOptions) (store.ObjectQuarantine, error) { //nolint:ireturn
+func (loose *Loose) BeginObjectQuarantine(_ store.ObjectQuarantineOptions) (store.ObjectQuarantine, error) {
tempName, tempRoot, err := createLooseQuarantineRoot(loose.root)
if err != nil {
return nil, err
diff --git a/object/store/packed/internal/ingest/basecache.go b/object/store/packed/internal/ingest/basecache.go
new file mode 100644
index 00000000..77419aa7
--- /dev/null
+++ b/object/store/packed/internal/ingest/basecache.go
@@ -0,0 +1,25 @@
+package ingest
+
+import (
+ "lindenii.org/go/furgit/internal/cache/clock"
+ "lindenii.org/go/furgit/object/typ"
+)
+
+const baseCacheMaxWeight = 96 << 20
+
+type baseCacheKey struct {
+ offset int
+}
+
+type cachedContent struct {
+ objectType typ.Type
+ content []byte
+}
+
+func newBaseCache(workers int) *clock.Clock[baseCacheKey, cachedContent] {
+ return clock.New(baseCacheMaxWeight*uint64(workers), baseContentWeight) //#nosec G115
+}
+
+func baseContentWeight(_ baseCacheKey, base cachedContent) uint64 {
+ return uint64(len(base.content)) + 32
+}
diff --git a/object/store/packed/internal/ingest/finalize.go b/object/store/packed/internal/ingest/finalize.go
index afed996c..c6b1e2c9 100644
--- a/object/store/packed/internal/ingest/finalize.go
+++ b/object/store/packed/internal/ingest/finalize.go
@@ -18,6 +18,11 @@ import (
// then links the pack, reverse index, and index
// to their content-addressed names.
func (ingestion *ingestion) finalize() (Result, error) {
+ err := ingestion.ctx.Err()
+ if err != nil {
+ return Result{}, fmt.Errorf("object/store/packed/internal/ingest: %w", err)
+ }
+
entries, positions, err := ingestion.indexEntries()
if err != nil {
return Result{}, err
@@ -53,37 +58,41 @@ func (ingestion *ingestion) finalize() (Result, error) {
return Result{}, err
}
+ objectCount, err := intconv.IntToUint32(len(ingestion.records))
+ if err != nil {
+ return Result{}, fmt.Errorf("object/store/packed/internal/ingest: %w", err)
+ }
+
base := "pack-" + ingestion.packHash.String()
packFinal := base + ".pack"
idxFinal := base + ".idx"
revFinal := base + ".rev"
bloomFinal := base + ".bloom"
- // Link the pack, reverse index, and Bloom filter before the index,
+ // Link the data files before the index,
// since the index is what publishes the pack to readers.
- err = ingestion.link(ingestion.packTmp, packFinal)
- if err != nil {
- return Result{}, err
+ artifacts := [...]struct{ tmp, final string }{
+ {ingestion.packTmp, packFinal},
+ {revTmp, revFinal},
+ {bloomTmp, bloomFinal},
+ {idxTmp, idxFinal},
}
- err = ingestion.link(revTmp, revFinal)
- if err != nil {
- return Result{}, err
- }
+ var created []string
- err = ingestion.link(bloomTmp, bloomFinal)
- if err != nil {
- return Result{}, err
- }
+ for _, artifact := range artifacts {
+ linked, err := ingestion.promote(artifact.tmp, artifact.final)
+ if err != nil {
+ for i := len(created) - 1; i >= 0; i-- {
+ _ = ingestion.root.Remove(created[i])
+ }
- err = ingestion.link(idxTmp, idxFinal)
- if err != nil {
- return Result{}, err
- }
+ return Result{}, err
+ }
- objectCount, err := intconv.IntToUint32(len(ingestion.records))
- if err != nil {
- return Result{}, fmt.Errorf("object/store/packed/internal/ingest: %w", err)
+ if linked {
+ created = append(created, artifact.final)
+ }
}
return Result{
@@ -184,15 +193,21 @@ func (ingestion *ingestion) writeTemp(prefix string, write func(io.Writer) error
return name, nil
}
-// link hard-links tmp to final,
-// treating an already-present destination as success.
-func (ingestion *ingestion) link(tmp, final string) error {
+// promote hard-links tmp to final and reports whether final was newly created.
+// A pre-existing final is treated as success; rollback must not remove it.
+func (ingestion *ingestion) promote(tmp, final string) (bool, error) {
err := ingestion.root.Link(tmp, final)
- if err != nil && !errors.Is(err, fs.ErrExist) {
- return fmt.Errorf("object/store/packed/internal/ingest: linking %q: %w", final, err)
- }
- _ = ingestion.root.Remove(tmp)
+ switch {
+ case err == nil:
+ _ = ingestion.root.Remove(tmp)
+
+ return true, nil
+ case errors.Is(err, fs.ErrExist):
+ _ = ingestion.root.Remove(tmp)
- return nil
+ return false, nil
+ default:
+ return false, fmt.Errorf("object/store/packed/internal/ingest: linking %q: %w", final, err)
+ }
}
diff --git a/object/store/packed/internal/ingest/ingest.go b/object/store/packed/internal/ingest/ingest.go
index 5422b4af..9b60af85 100644
--- a/object/store/packed/internal/ingest/ingest.go
+++ b/object/store/packed/internal/ingest/ingest.go
@@ -2,22 +2,29 @@ package ingest
import (
"bytes"
+ "context"
"crypto/rand"
"errors"
"fmt"
"io"
"io/fs"
"os"
+ "runtime"
+ "sync/atomic"
+ "lindenii.org/go/furgit/internal/cache/clock"
"lindenii.org/go/furgit/internal/format/packfile"
"lindenii.org/go/furgit/object/id"
"lindenii.org/go/furgit/object/store"
+ "lindenii.org/go/lgo/sync"
)
var errTempNamesExhausted = errors.New("object/store/packed/internal/ingest: exhausted temporary file names")
// ingestion holds the state for one WritePack call.
type ingestion struct {
+ ctx context.Context //nolint:containedctx
+
// root is the destination objects/pack directory.
root *os.Root
@@ -47,7 +54,12 @@ type ingestion struct {
// byOffset maps an entry offset to its record index,
// and byOID maps a resolved object ID to its record index.
byOffset map[int]int
- byOID map[id.ObjectID]int
+ byOID sync.Map[id.ObjectID, int]
+
+ baseCache *clock.Clock[baseCacheKey, cachedContent]
+
+ // workers is the delta-resolution concurrency.
+ workers int
// headerCount is the object count declared by the pack header.
headerCount int
@@ -55,8 +67,8 @@ type ingestion struct {
// deltaCount counts delta records, accumulated during scanning.
deltaCount int
- // deltasResolved counts resolved delta records, for progress.
- deltasResolved int
+ // deltasResolved counts resolved delta records.
+ deltasResolved atomic.Int64
// packHash is the final pack trailer hash.
packHash id.ObjectID
@@ -81,7 +93,7 @@ type ingestion struct {
// The pack must be the last thing the peer sends before that response:
// any bytes arriving immediately after the trailer
// are rejected as a malformed pack.
-func WritePack(root *os.Root, objectFormat id.ObjectFormat, src io.Reader, opts store.PackWriteOptions) (Result, error) {
+func WritePack(ctx context.Context, root *os.Root, objectFormat id.ObjectFormat, src io.Reader, opts store.PackWriteOptions) (Result, error) {
if objectFormat.Size() == 0 {
return Result{}, id.ErrInvalidObjectFormat
}
@@ -91,24 +103,27 @@ func WritePack(root *os.Root, objectFormat id.ObjectFormat, src io.Reader, opts
return Result{}, err
}
+ workers := runtime.GOMAXPROCS(0)
+
ingestion := &ingestion{
- root: root,
- objectFormat: objectFormat,
- opts: opts,
- src: src,
- packFile: nil,
- packTmp: "",
- temps: nil,
- scanner: nil,
- records: nil,
- byOffset: make(map[int]int),
- byOID: make(map[id.ObjectID]int),
- headerCount: count,
- deltaCount: 0,
- deltasResolved: 0,
- packHash: id.ObjectID{},
- thinFixed: false,
- committed: false,
+ ctx: ctx,
+ root: root,
+ objectFormat: objectFormat,
+ opts: opts,
+ src: src,
+ packFile: nil,
+ packTmp: "",
+ temps: nil,
+ scanner: nil,
+ records: nil,
+ byOffset: make(map[int]int),
+ baseCache: newBaseCache(workers),
+ workers: workers,
+ headerCount: count,
+ deltaCount: 0,
+ packHash: id.ObjectID{},
+ thinFixed: false,
+ committed: false,
}
defer ingestion.cleanup()
diff --git a/object/store/packed/internal/ingest/record.go b/object/store/packed/internal/ingest/record.go
index 69101293..4031a246 100644
--- a/object/store/packed/internal/ingest/record.go
+++ b/object/store/packed/internal/ingest/record.go
@@ -37,15 +37,11 @@ type record struct {
// baseOID is the base object ID for a ref-delta.
baseOID id.ObjectID
- // objectType is the resolved object type,
- // meaningful once resolved is true.
- objectType packfile.EntryType
-
// oid is the resolved object ID,
// meaningful once resolved is true.
oid id.ObjectID
- // resolved reports whether oid and objectType are final.
+ // resolved reports whether oid is final.
resolved bool
}
diff --git a/object/store/packed/internal/ingest/resolve.go b/object/store/packed/internal/ingest/resolve.go
index 77b0fa0f..dd26cd47 100644
--- a/object/store/packed/internal/ingest/resolve.go
+++ b/object/store/packed/internal/ingest/resolve.go
@@ -3,6 +3,7 @@ package ingest
import (
"fmt"
"io"
+ "sync"
"lindenii.org/go/furgit/internal/compress/zlib"
"lindenii.org/go/furgit/internal/format/packfile"
@@ -10,6 +11,8 @@ import (
"lindenii.org/go/furgit/internal/progress"
"lindenii.org/go/furgit/object/header"
"lindenii.org/go/furgit/object/id"
+ "lindenii.org/go/furgit/object/store"
+ "lindenii.org/go/furgit/object/typ"
)
// adjacency maps each resolvable base to its delta children:
@@ -83,108 +86,311 @@ func (ingestion *ingestion) buildAdjacency() adjacency {
return out
}
-// resolveFrom resolves the delta subtree rooted at each resolved record.
+// item is a delta record awaiting resolution, with its delta-chain depth.
+type item struct {
+ index int
+ depth int
+}
+
+// resolver resolves deltas concurrently over a shared LIFO work stack.
+//
+// Each item is one delta child;
+// a worker materializes its base from the cache, re-deriving on a miss,
+// resolves the child,
+// and pushes the child's own delta children.
+// Workers park while the stack is empty but others are still working,
+// and exit once it is empty and none are.
+type resolver struct {
+ ingestion *ingestion
+ adjacency adjacency
+ meter *progress.Meter
+
+ mu sync.Mutex
+ cond *sync.Cond
+ stack []item
+ active int
+ firstErr error
+}
+
func (ingestion *ingestion) resolveFrom(roots []int, adjacency adjacency, meter *progress.Meter) error {
+ var seed []item
+
for _, root := range roots {
- content, err := ingestion.inflateRecord(root)
- if err != nil {
- return err
+ rec := &ingestion.records[root]
+ for _, group := range [2][]int{adjacency.byOffset[rec.offset], adjacency.byOID[rec.oid]} {
+ for _, child := range group {
+ seed = append(seed, item{index: child, depth: 1})
+ }
}
+ }
- err = ingestion.resolveSubtree(root, content, ingestion.records[root].objectType, 0, adjacency, meter)
- if err != nil {
- return err
- }
+ if len(seed) == 0 {
+ return nil
}
- return nil
+ res := &resolver{
+ ingestion: ingestion,
+ adjacency: adjacency,
+ meter: meter,
+ stack: seed,
+ }
+ res.cond = sync.NewCond(&res.mu)
+
+ return res.run(ingestion.workers)
}
-// resolveSubtree resolves every delta child of one resolved record at depth,
-// holding the record's content as the base for its children.
-func (ingestion *ingestion) resolveSubtree(
- index int,
- content []byte,
- objectType packfile.EntryType,
- depth int,
- adjacency adjacency,
- meter *progress.Meter,
-) error {
- rec := &ingestion.records[index]
+func (res *resolver) run(workers int) error {
+ if workers <= 1 {
+ res.worker()
- for _, child := range adjacency.byOffset[rec.offset] {
- err := ingestion.resolveChild(child, content, objectType, depth+1, adjacency, meter)
- if err != nil {
- return err
+ return res.firstErr
+ }
+
+ var wg sync.WaitGroup
+
+ for range workers {
+ wg.Go(func() {
+ res.worker()
+ })
+ }
+
+ wg.Wait()
+
+ return res.firstErr
+}
+
+func (res *resolver) worker() {
+ for {
+ res.mu.Lock()
+
+ for len(res.stack) == 0 && res.active > 0 && res.firstErr == nil {
+ res.cond.Wait()
+ }
+
+ if res.firstErr != nil || len(res.stack) == 0 {
+ res.mu.Unlock()
+
+ return
+ }
+
+ it := res.stack[len(res.stack)-1]
+ res.stack = res.stack[:len(res.stack)-1]
+ res.active++
+ res.mu.Unlock()
+
+ kids, err := res.process(it)
+
+ res.mu.Lock()
+ res.active--
+
+ if err != nil && res.firstErr == nil {
+ res.firstErr = err
+ }
+
+ if res.firstErr == nil {
+ res.stack = append(res.stack, kids...)
}
+
+ if res.firstErr != nil || len(kids) > 0 || (res.active == 0 && len(res.stack) == 0) {
+ res.cond.Broadcast()
+ }
+
+ res.mu.Unlock()
}
+}
- for _, child := range adjacency.byOID[rec.oid] {
- err := ingestion.resolveChild(child, content, objectType, depth+1, adjacency, meter)
- if err != nil {
- return err
+// process resolves one delta child and returns its own delta children.
+func (res *resolver) process(it item) ([]item, error) {
+ err := res.ingestion.ctx.Err()
+ if err != nil {
+ return nil, fmt.Errorf("object/store/packed/internal/ingest: %w", err)
+ }
+
+ rec := &res.ingestion.records[it.index]
+
+ parent, ok := res.ingestion.baseRecordIndex(rec)
+ if !ok {
+ return nil, fmt.Errorf("%w: entry at %d: base unavailable while resolving", ErrMalformedPack, rec.offset)
+ }
+
+ baseType, baseContent, err := res.ingestion.materialize(parent)
+ if err != nil {
+ return nil, err
+ }
+
+ err = res.ingestion.resolveOneChild(it.index, baseType, baseContent, res.meter)
+ if err != nil {
+ return nil, err
+ }
+
+ return res.childItems(it.index, it.depth+1)
+}
+
+// childItems returns the delta children of a just-resolved record at depth.
+func (res *resolver) childItems(index, depth int) ([]item, error) {
+ rec := &res.ingestion.records[index]
+
+ var kids []item
+
+ for _, group := range [2][]int{res.adjacency.byOffset[rec.offset], res.adjacency.byOID[rec.oid]} {
+ for _, child := range group {
+ if depth > delta.MaxChainDepth {
+ return nil, fmt.Errorf("%w: entry at %d: delta chain too deep", ErrMalformedPack, res.ingestion.records[child].offset)
+ }
+
+ kids = append(kids, item{index: child, depth: depth})
}
}
- return nil
+ return kids, nil
}
-// resolveChild applies one delta record at depth against its base content,
-// finalizes the record, and recurses into its own children.
-func (ingestion *ingestion) resolveChild(
- index int,
- baseContent []byte,
- baseType packfile.EntryType,
- depth int,
- adjacency adjacency,
- meter *progress.Meter,
-) error {
+func (ingestion *ingestion) resolveOneChild(index int, baseType typ.Type, baseContent []byte, meter *progress.Meter) error {
rec := &ingestion.records[index]
- if rec.resolved {
- return nil
+
+ content, err := ingestion.applyDelta(index, baseContent)
+ if err != nil {
+ return err
+ }
+
+ oid, err := ingestion.hashObject(baseType, content)
+ if err != nil {
+ return err
+ }
+
+ rec.oid = oid
+ rec.resolved = true
+
+ ingestion.byOID.Store(oid, index)
+ ingestion.baseCache.Add(baseCacheKey{offset: rec.offset}, cachedContent{objectType: baseType, content: content})
+
+ ingestion.deltasResolved.Add(1)
+ meter.Add(1, 0)
+
+ return nil
+}
+
+// materialize returns the inflated content of an already-resolved record,
+// from the base cache,
+// or re-derived from the nearest cached or base ancestor on a miss.
+func (ingestion *ingestion) materialize(index int) (typ.Type, []byte, error) {
+ var (
+ zero typ.Type
+ chain []int
+ base []byte
+ baseType typ.Type
+ )
+
+ cur := index
+
+ for {
+ rec := &ingestion.records[cur]
+
+ if cached, ok := ingestion.baseCache.Get(baseCacheKey{offset: rec.offset}); ok {
+ base = cached.content
+ baseType = cached.objectType
+
+ break
+ }
+
+ if rec.packedType.IsBase() {
+ objectType, err := rec.packedType.ObjectType()
+ if err != nil {
+ return zero, nil, fmt.Errorf("object/store/packed/internal/ingest: %w", err)
+ }
+
+ content, err := ingestion.inflateRecord(cur)
+ if err != nil {
+ return zero, nil, err
+ }
+
+ base = content
+ baseType = objectType
+
+ break
+ }
+
+ if len(chain) >= delta.MaxChainDepth {
+ return zero, nil, fmt.Errorf("%w: entry at %d: delta chain too deep", ErrMalformedPack, rec.offset)
+ }
+
+ chain = append(chain, cur)
+
+ next, ok := ingestion.baseRecordIndex(rec)
+ if !ok {
+ return zero, nil, fmt.Errorf("%w: entry at %d: base unavailable while reconstructing", ErrMalformedPack, rec.offset)
+ }
+
+ cur = next
}
- if depth > delta.MaxChainDepth {
- return fmt.Errorf("%w: entry at %d: delta chain too deep", ErrMalformedPack, rec.offset)
+ for i := len(chain) - 1; i >= 0; i-- {
+ content, err := ingestion.applyDelta(chain[i], base)
+ if err != nil {
+ return zero, nil, err
+ }
+
+ ingestion.baseCache.Add(baseCacheKey{offset: ingestion.records[chain[i]].offset}, cachedContent{objectType: baseType, content: content})
+
+ base = content
}
+ return baseType, base, nil
+}
+
+func (ingestion *ingestion) applyDelta(index int, baseContent []byte) ([]byte, error) {
+ rec := &ingestion.records[index]
+
deltaPayload, err := ingestion.inflateRecord(index)
if err != nil {
- return err
+ return nil, err
}
baseSize, resultSize, _, err := delta.ParseHeaderSizes(deltaPayload)
if err != nil {
- return fmt.Errorf("%w: entry at %d: %w", ErrMalformedPack, rec.offset, err)
+ return nil, fmt.Errorf("%w: entry at %d: %w", ErrMalformedPack, rec.offset, err)
+ }
+
+ limit := ingestion.opts.MaxObjectSize
+ if limit > 0 && resultSize > uint64(limit) {
+ return nil, fmt.Errorf("%w: entry at %d: result size %d exceeds limit %d", store.ErrObjectTooLarge, rec.offset, resultSize, limit)
}
if baseSize != uint64(len(baseContent)) {
- return fmt.Errorf("%w: entry at %d: delta base size mismatch", ErrMalformedPack, rec.offset)
+ return nil, fmt.Errorf("%w: entry at %d: delta base size mismatch", ErrMalformedPack, rec.offset)
}
content, err := delta.Apply(baseContent, deltaPayload)
if err != nil {
- return fmt.Errorf("%w: entry at %d: %w", ErrMalformedPack, rec.offset, err)
+ return nil, fmt.Errorf("%w: entry at %d: %w", ErrMalformedPack, rec.offset, err)
}
if uint64(len(content)) != resultSize {
- return fmt.Errorf("%w: entry at %d: delta result size mismatch", ErrMalformedPack, rec.offset)
+ return nil, fmt.Errorf("%w: entry at %d: delta result size mismatch", ErrMalformedPack, rec.offset)
}
- oid, err := ingestion.hashObject(baseType, content)
- if err != nil {
- return err
- }
+ return content, nil
+}
- rec.objectType = baseType
- rec.oid = oid
- rec.resolved = true
- ingestion.byOID[oid] = index
+func (ingestion *ingestion) baseRecordIndex(rec *record) (int, bool) {
+ switch rec.packedType {
+ case packfile.EntryTypeOfsDelta:
+ index, ok := ingestion.byOffset[rec.baseOffset]
+
+ return index, ok
+ case packfile.EntryTypeRefDelta:
+ index, ok := ingestion.byOID.Load(rec.baseOID)
- ingestion.deltasResolved++
- meter.Set(ingestion.deltasResolved, 0)
+ return index, ok
+ case packfile.EntryTypeInvalid,
+ packfile.EntryTypeCommit,
+ packfile.EntryTypeTree,
+ packfile.EntryTypeBlob,
+ packfile.EntryTypeTag,
+ packfile.EntryTypeFuture:
+ }
- return ingestion.resolveSubtree(index, content, baseType, depth, adjacency, meter)
+ return 0, false
}
// inflateRecord inflates one record's payload from the temporary pack file.
@@ -213,20 +419,15 @@ func (ingestion *ingestion) inflateRecord(index int) ([]byte, error) {
}
// hashObject computes the object ID of one resolved object.
-func (ingestion *ingestion) hashObject(objectType packfile.EntryType, content []byte) (id.ObjectID, error) {
+func (ingestion *ingestion) hashObject(objectType typ.Type, content []byte) (id.ObjectID, error) {
var zero id.ObjectID
- ty, err := objectType.ObjectType()
- if err != nil {
- return zero, fmt.Errorf("object/store/packed/internal/ingest: %w", err)
- }
-
hashImpl, err := ingestion.objectFormat.New()
if err != nil {
return zero, fmt.Errorf("object/store/packed/internal/ingest: %w", err)
}
- _, _ = hashImpl.Write(header.Append(nil, ty, len(content)))
+ _, _ = hashImpl.Write(header.Append(nil, objectType, len(content)))
_, _ = hashImpl.Write(content)
oid, err := ingestion.objectFormat.FromBytes(hashImpl.Sum(nil))
@@ -261,7 +462,7 @@ func (ingestion *ingestion) countDeltas() int {
// so the unresolved records are exactly the unresolved deltas:
// the delta records minus those already resolved.
func (ingestion *ingestion) countUnresolved() int {
- return ingestion.deltaCount - ingestion.deltasResolved
+ return ingestion.deltaCount - int(ingestion.deltasResolved.Load())
}
// unresolvedExternalBases returns the unique base object IDs
@@ -270,7 +471,7 @@ func (ingestion *ingestion) countUnresolved() int {
func (ingestion *ingestion) unresolvedExternalBases() []id.ObjectID {
seen := make(map[id.ObjectID]struct{})
- out := make([]id.ObjectID, 0, ingestion.deltaCount-ingestion.deltasResolved)
+ out := make([]id.ObjectID, 0, ingestion.deltaCount-int(ingestion.deltasResolved.Load()))
for index := range ingestion.records {
rec := &ingestion.records[index]
@@ -278,7 +479,7 @@ func (ingestion *ingestion) unresolvedExternalBases() []id.ObjectID {
continue
}
- if _, ok := ingestion.byOID[rec.baseOID]; ok {
+ if _, ok := ingestion.byOID.Load(rec.baseOID); ok {
continue
}
diff --git a/object/store/packed/internal/ingest/scan.go b/object/store/packed/internal/ingest/scan.go
index 6b3b73b7..2cb5c135 100644
--- a/object/store/packed/internal/ingest/scan.go
+++ b/object/store/packed/internal/ingest/scan.go
@@ -13,6 +13,7 @@ import (
"lindenii.org/go/furgit/internal/progress"
"lindenii.org/go/furgit/object/header"
"lindenii.org/go/furgit/object/id"
+ "lindenii.org/go/furgit/object/store"
"lindenii.org/go/lgo/intconv"
)
@@ -294,13 +295,22 @@ func (ingestion *ingestion) streamAndScan() error {
Throughput: true,
})
- for done := range ingestion.headerCount {
- err := ingestion.scanEntry(ingestion.scanner.consumed)
+ prevConsumed := ingestion.scanner.consumed
+
+ for range ingestion.headerCount {
+ err := ingestion.ctx.Err()
+ if err != nil {
+ return fmt.Errorf("object/store/packed/internal/ingest: %w", err)
+ }
+
+ err = ingestion.scanEntry(ingestion.scanner.consumed)
if err != nil {
return err
}
- meter.Set(done+1, ingestion.scanner.consumed)
+ consumed := ingestion.scanner.consumed
+ meter.Add(1, int64(consumed-prevConsumed))
+ prevConsumed = consumed
}
meter.Stop("done")
@@ -345,7 +355,6 @@ func (ingestion *ingestion) scanEntry(start int) error {
rec.crc32 = ingestion.scanner.endCRC()
if rec.packedType.IsBase() {
- rec.objectType = rec.packedType
rec.oid = oid
rec.resolved = true
} else {
@@ -357,7 +366,7 @@ func (ingestion *ingestion) scanEntry(start int) error {
ingestion.byOffset[rec.offset] = index
if rec.resolved {
- ingestion.byOID[rec.oid] = index
+ ingestion.byOID.Store(rec.oid, index)
}
return nil
@@ -384,6 +393,11 @@ func (ingestion *ingestion) scanHeader(start int) (record, error) {
return rec, fmt.Errorf("%w: entry at %d: declared size overflows int: %w", ErrMalformedPack, start, err)
}
+ limit := ingestion.opts.MaxObjectSize
+ if limit > 0 && declaredSize > limit {
+ return rec, fmt.Errorf("%w: entry at %d: declared size %d exceeds limit %d", store.ErrObjectTooLarge, start, declaredSize, limit)
+ }
+
rec.packedType = entryHeader.Type
rec.declaredSize = declaredSize
rec.headerLen = entryHeader.HeaderLen
diff --git a/object/store/packed/internal/ingest/thin.go b/object/store/packed/internal/ingest/thin.go
index fa125f2f..15773a56 100644
--- a/object/store/packed/internal/ingest/thin.go
+++ b/object/store/packed/internal/ingest/thin.go
@@ -36,6 +36,11 @@ func (ingestion *ingestion) fixThin(external []id.ObjectID, adjacency adjacency,
appended := make([]int, 0, len(external))
for _, baseOID := range external {
+ err := ingestion.ctx.Err()
+ if err != nil {
+ return fmt.Errorf("object/store/packed/internal/ingest: %w", err)
+ }
+
ty, content, err := ingestion.opts.ThinBase.ReadBytesContent(baseOID)
if errors.Is(err, store.ErrObjectNotFound) {
continue
@@ -86,7 +91,7 @@ func (ingestion *ingestion) appendBaseObject(objectID id.ObjectID, objectType ty
return 0, fmt.Errorf("object/store/packed/internal/ingest: %w", err)
}
- computed, err := ingestion.hashObject(entryType, content)
+ computed, err := ingestion.hashObject(objectType, content)
if err != nil {
return 0, err
}
@@ -138,7 +143,6 @@ func (ingestion *ingestion) appendBaseObject(objectID id.ObjectID, objectType ty
declaredSize: len(content),
baseOffset: 0,
baseOID: id.ObjectID{},
- objectType: entryType,
oid: objectID,
resolved: true,
}
@@ -146,7 +150,7 @@ func (ingestion *ingestion) appendBaseObject(objectID id.ObjectID, objectType ty
index := len(ingestion.records)
ingestion.records = append(ingestion.records, rec)
ingestion.byOffset[start] = index
- ingestion.byOID[objectID] = index
+ ingestion.byOID.Store(objectID, index)
return index, nil
}
diff --git a/object/store/packed/internal/ingest/writepack_test.go b/object/store/packed/internal/ingest/writepack_test.go
index adc0ba35..b2f4d2b8 100644
--- a/object/store/packed/internal/ingest/writepack_test.go
+++ b/object/store/packed/internal/ingest/writepack_test.go
@@ -2,6 +2,7 @@ package ingest_test
import (
"bytes"
+ "context"
"errors"
"io"
"os"
@@ -256,7 +257,7 @@ func TestWritePackIdempotent(t *testing.T) {
t.Cleanup(func() { _ = root.Close() })
- first, err := ingest.WritePack(root, objectFormat, bytes.NewReader(stream), store.PackWriteOptions{
+ first, err := ingest.WritePack(t.Context(), root, objectFormat, bytes.NewReader(stream), store.PackWriteOptions{
ThinBase: nil,
Progress: nil,
})
@@ -264,7 +265,7 @@ func TestWritePackIdempotent(t *testing.T) {
t.Fatalf("first WritePack: %v", err)
}
- second, err := ingest.WritePack(root, objectFormat, bytes.NewReader(stream), store.PackWriteOptions{
+ second, err := ingest.WritePack(t.Context(), root, objectFormat, bytes.NewReader(stream), store.PackWriteOptions{
ThinBase: nil,
Progress: nil,
})
@@ -305,7 +306,7 @@ func writePack(
t.Cleanup(func() { _ = root.Close() })
- result, err := ingest.WritePack(root, objectFormat, src, opts)
+ result, err := ingest.WritePack(t.Context(), root, objectFormat, src, opts)
if err != nil {
t.Fatalf("WritePack: %v", err)
}
@@ -355,7 +356,7 @@ func TestWritePackThinWithoutBase(t *testing.T) {
repo, seeded := seedHistory(t, objectFormat)
stream := thinStream(t, repo, seeded)
- _, err := ingest.WritePack(freshRoot(t), objectFormat, bytes.NewReader(stream), store.PackWriteOptions{
+ _, err := ingest.WritePack(t.Context(), freshRoot(t), objectFormat, bytes.NewReader(stream), store.PackWriteOptions{
ThinBase: nil,
Progress: nil,
})
@@ -381,7 +382,7 @@ func TestWritePackThinMissingBase(t *testing.T) {
emptyBase := emptyStore(t, objectFormat)
stream := thinStream(t, repo, seeded)
- _, err := ingest.WritePack(freshRoot(t), objectFormat, bytes.NewReader(stream), store.PackWriteOptions{
+ _, err := ingest.WritePack(t.Context(), freshRoot(t), objectFormat, bytes.NewReader(stream), store.PackWriteOptions{
ThinBase: emptyBase,
Progress: nil,
})
@@ -398,6 +399,118 @@ func TestWritePackThinMissingBase(t *testing.T) {
}
}
+// TestWritePackContextCancelled verifies that a cancelled context
+// aborts ingestion and publishes no artifacts.
+func TestWritePackContextCancelled(t *testing.T) {
+ t.Parallel()
+
+ for _, objectFormat := range id.SupportedObjectFormats() {
+ t.Run(objectFormat.String(), func(t *testing.T) {
+ t.Parallel()
+
+ repo, seeded := seedHistory(t, objectFormat)
+
+ gitPrefix, err := repo.PackObjects(t, seeded.All(), testgit.PackObjectsOptions{
+ RevIndex: false,
+ Revs: false,
+ Exclude: nil,
+ })
+ if err != nil {
+ t.Fatalf("PackObjects: %v", err)
+ }
+
+ stream, err := os.ReadFile(gitPrefix + ".pack") //nolint:gosec
+ if err != nil {
+ t.Fatalf("ReadFile pack: %v", err)
+ }
+
+ ctx, cancel := context.WithCancel(t.Context())
+ cancel()
+
+ dir := t.TempDir()
+
+ root, err := os.OpenRoot(dir)
+ if err != nil {
+ t.Fatalf("OpenRoot: %v", err)
+ }
+
+ t.Cleanup(func() { _ = root.Close() })
+
+ _, err = ingest.WritePack(ctx, root, objectFormat, bytes.NewReader(stream), store.PackWriteOptions{
+ ThinBase: nil,
+ Progress: nil,
+ })
+ if !errors.Is(err, context.Canceled) {
+ t.Fatalf("err = %v, want context.Canceled", err)
+ }
+
+ entries, err := os.ReadDir(dir)
+ if err != nil {
+ t.Fatalf("ReadDir: %v", err)
+ }
+
+ if len(entries) != 0 {
+ t.Fatalf("cancelled ingestion left %d files behind", len(entries))
+ }
+ })
+ }
+}
+
+// TestWritePackObjectTooLarge verifies that an object exceeding MaxObjectSize
+// is rejected and no artifacts are published.
+func TestWritePackObjectTooLarge(t *testing.T) {
+ t.Parallel()
+
+ for _, objectFormat := range id.SupportedObjectFormats() {
+ t.Run(objectFormat.String(), func(t *testing.T) {
+ t.Parallel()
+
+ repo, seeded := seedHistory(t, objectFormat)
+
+ gitPrefix, err := repo.PackObjects(t, seeded.All(), testgit.PackObjectsOptions{
+ RevIndex: false,
+ Revs: false,
+ Exclude: nil,
+ })
+ if err != nil {
+ t.Fatalf("PackObjects: %v", err)
+ }
+
+ stream, err := os.ReadFile(gitPrefix + ".pack") //nolint:gosec
+ if err != nil {
+ t.Fatalf("ReadFile pack: %v", err)
+ }
+
+ dir := t.TempDir()
+
+ root, err := os.OpenRoot(dir)
+ if err != nil {
+ t.Fatalf("OpenRoot: %v", err)
+ }
+
+ t.Cleanup(func() { _ = root.Close() })
+
+ _, err = ingest.WritePack(t.Context(), root, objectFormat, bytes.NewReader(stream), store.PackWriteOptions{
+ ThinBase: nil,
+ Progress: nil,
+ MaxObjectSize: 1,
+ })
+ if !errors.Is(err, store.ErrObjectTooLarge) {
+ t.Fatalf("err = %v, want ErrObjectTooLarge", err)
+ }
+
+ entries, err := os.ReadDir(dir)
+ if err != nil {
+ t.Fatalf("ReadDir: %v", err)
+ }
+
+ if len(entries) != 0 {
+ t.Fatalf("rejected ingestion left %d files behind", len(entries))
+ }
+ })
+ }
+}
+
// seedHistory creates one repository with a seeded history.
func seedHistory(t *testing.T, objectFormat id.ObjectFormat) (*testgit.Repo, testgit.Seeded) {
t.Helper()
diff --git a/object/store/packed/quarantine.go b/object/store/packed/quarantine.go
index 977a9543..6f6a8c18 100644
--- a/object/store/packed/quarantine.go
+++ b/object/store/packed/quarantine.go
@@ -95,29 +95,46 @@ func (quarantine *packQuarantine) promoteAll() error {
return packPromotionPriority(left.Name()) - packPromotionPriority(right.Name())
})
+ var created []string
+
for _, entry := range entries {
- err := quarantine.promoteFile(entry.Name())
+ linked, err := quarantine.promoteFile(entry.Name())
if err != nil {
+ for i := len(created) - 1; i >= 0; i-- {
+ _ = quarantine.parent.root.Remove(created[i])
+ }
+
return err
}
+
+ if linked {
+ created = append(created, entry.Name())
+ }
}
return nil
}
-// promoteFile links one quarantined artifact into the parent store,
-// treating an already-present destination as success.
-func (quarantine *packQuarantine) promoteFile(name string) error {
+// promoteFile links one quarantined artifact into the parent store
+// and reports whether the destination was newly created.
+// A pre-existing destination is treated as success; rollback must not remove it.
+func (quarantine *packQuarantine) promoteFile(name string) (bool, error) {
src := quarantine.tempName + "/" + name
err := quarantine.parent.root.Link(src, name)
- if err != nil && !errors.Is(err, fs.ErrExist) {
- return fmt.Errorf("object/store/packed: promoting %q: %w", name, err)
- }
- _ = quarantine.parent.root.Remove(src)
+ switch {
+ case err == nil:
+ _ = quarantine.parent.root.Remove(src)
- return nil
+ return true, nil
+ case errors.Is(err, fs.ErrExist):
+ _ = quarantine.parent.root.Remove(src)
+
+ return false, nil
+ default:
+ return false, fmt.Errorf("object/store/packed: promoting %q: %w", name, err)
+ }
}
// createPackQuarantineRoot creates a private quarantine directory beneath parent
diff --git a/object/store/packed/writer.go b/object/store/packed/writer.go
index 59309c24..6476cc42 100644
--- a/object/store/packed/writer.go
+++ b/object/store/packed/writer.go
@@ -1,6 +1,7 @@
package packed
import (
+ "context"
"fmt"
"io"
@@ -23,8 +24,8 @@ var _ store.PackWriter = (*Packed)(nil)
// The pack must be the last thing the peer sends before that response:
// any bytes arriving immediately after the trailer
// are rejected as a malformed pack.
-func (packed *Packed) WritePack(src io.Reader, opts store.PackWriteOptions) error {
- _, err := ingest.WritePack(packed.root, packed.objectFormat, src, opts)
+func (packed *Packed) WritePack(ctx context.Context, src io.Reader, opts store.PackWriteOptions) error {
+ _, err := ingest.WritePack(ctx, packed.root, packed.objectFormat, src, opts)
if err != nil {
return err //nolint:wrapcheck
}
diff --git a/object/store/packed/writer_test.go b/object/store/packed/writer_test.go
index 8227caa7..d668647b 100644
--- a/object/store/packed/writer_test.go
+++ b/object/store/packed/writer_test.go
@@ -42,7 +42,7 @@ func TestWritePack(t *testing.T) {
packedStore := openEmptyStore(t, objectFormat)
- err = packedStore.WritePack(bytes.NewReader(stream), store.PackWriteOptions{
+ err = packedStore.WritePack(t.Context(), bytes.NewReader(stream), store.PackWriteOptions{
ThinBase: nil,
Progress: nil,
})
diff --git a/object/store/writer.go b/object/store/writer.go
index d83eec6a..0437505d 100644
--- a/object/store/writer.go
+++ b/object/store/writer.go
@@ -1,6 +1,7 @@
package store
import (
+ "context"
"errors"
"io"
@@ -12,6 +13,10 @@ import (
// ErrInvalidObject indicates a malformed object passed to a write.
var ErrInvalidObject = errors.New("object/store: invalid object")
+// ErrObjectTooLarge indicates that an object exceeds
+// the size limit configured for the write.
+var ErrObjectTooLarge = errors.New("object/store: object too large")
+
// ObjectWriter writes individual Git objects.
type ObjectWriter interface {
// WriteBytesFull writes one full serialized object byte slice as "type size\x00content".
@@ -32,7 +37,7 @@ type PackWriter interface {
// WritePack ingests one pack stream,
// such that the objects contained therein
// become available in the relevant store.
- WritePack(src io.Reader, opts PackWriteOptions) error
+ WritePack(ctx context.Context, src io.Reader, opts PackWriteOptions) error
}
// PackWriteOptions controls one pack write operation.
@@ -65,4 +70,11 @@ type PackWriteOptions struct {
//
// When nil, no progress output is emitted.
Progress iowrap.WriteFlusher
+
+ // MaxObjectSize rejects ingestion of any object
+ // whose declared inflated size or delta result size exceeds it,
+ // bounding the memory spent reconstructing a single object.
+ //
+ // Zero or negative means no limit.
+ MaxObjectSize int
}
diff --git a/object/stored/stored.go b/object/stored/stored.go
index 68a7bfd0..99c497ee 100644
--- a/object/stored/stored.go
+++ b/object/stored/stored.go
@@ -18,7 +18,7 @@ func New[T object.Object](id id.ObjectID, obj T) *Stored[T] {
}
// Object returns the wrapped object as itself.
-func (stored *Stored[T]) Object() T { //nolint:ireturn
+func (stored *Stored[T]) Object() T {
return stored.obj
}
generated by cgit v1.3.1-10-gc9f91 (git 2.54.0) at 2026-07-06 00:03:26 +0000