diff --git a/repository/getblobs.go b/repository/getblobs.go new file mode 100644 index 00000000..e8e835d7 --- /dev/null +++ b/repository/getblobs.go @@ -0,0 +1,158 @@ +package repository + +import ( + "bytes" + "cmp" + "context" + "iter" + "slices" + "time" + + "github.com/PlakarKorp/kloset/objects" + "github.com/PlakarKorp/kloset/repository/state" + "github.com/PlakarKorp/kloset/resources" +) + +type BlobReq struct { + Type resources.Type + MAC objects.MAC +} + +// Output type of GetBlobs, embedding the requested blob (mac+type), and its +// associated data. If there was an error data is nil and the error is in the +// second part of the iterator. +type BlobResp struct { + BlobReq + Data []byte +} + +type GetBlobsOpts struct { +} + +const ( + // Default maximum range get size, 4 MB + maxSize = 4 << 20 +) + +type blobLocation struct { + BlobReq + state.Location +} + +type rangeRead struct { + loc state.Location // Actual range of data to get + blobs []blobLocation // blobs covered by that range. +} + +func (r *Repository) GetBlobs(ctx context.Context, blobs []BlobReq, opts *GetBlobsOpts) iter.Seq2[BlobResp, error] { + return func(yield func(BlobResp, error) bool) { + t0 := time.Now() + + locs := make([]blobLocation, 0, len(blobs)) + seen := make(map[BlobReq]struct{}) + + for _, b := range blobs { + if _, ok := seen[b]; ok { + continue + } + + seen[b] = struct{}{} + + loc, ok, err := r.state.GetSubpartForBlob(b.Type, b.MAC) + if err == nil && !ok { + err = ErrBlobNotFound + } + + if err != nil { + if !yield(BlobResp{b, nil}, err) { + return + } + + continue + } + + locs = append(locs, blobLocation{b, loc}) + } + + plan := readPlanner(locs, maxSize, 0.0) + defer func() { + r.Logger().Trace("repository", "GetBlobs(%d blobs): %d reads: %s", + len(blobs), len(plan), time.Since(t0)) + }() + + for _, g := range plan { + data, err := r.GetPackfileRange(g.loc) + + for _, b := range g.blobs { + if err != nil { + if !yield(BlobResp{b.BlobReq, nil}, err) { + return + } + + continue + } + + blobStart := b.Offset - g.loc.Offset + blobBytes := data[blobStart : blobStart+uint64(b.Length)] + blobData, err := r.decodeBuffer(blobBytes) + if err != nil { + if !yield(BlobResp{b.BlobReq, nil}, err) { + return + } + + continue + } + + if !yield(BlobResp{b.BlobReq, blobData}, nil) { + return + } + } + } + } +} + +// readPlanner tries to merge reads according to the policy given. +func readPlanner(locs []blobLocation, maxRdSize uint32, _ float64) []*rangeRead { + if len(locs) == 0 { + return nil + } + + // First sort by packfile then offset. + slices.SortFunc(locs, func(a, b blobLocation) int { + if c := bytes.Compare(a.Packfile[:], b.Packfile[:]); c == 0 { + return cmp.Compare(a.Offset, b.Offset) + } else { + return c + } + }) + + plan := make([]*rangeRead, 0) + + // append the first blob to simplify the following loop + rr := &rangeRead{ + loc: locs[0].Location, + blobs: []blobLocation{locs[0]}, + } + plan = append(plan, rr) + + for _, l := range locs[1:] { + curMerge := plan[len(plan)-1] + + if curMerge.loc.Packfile == l.Packfile { + size := (l.Offset + uint64(l.Length)) - curMerge.loc.Offset + if size <= uint64(maxRdSize) { + curMerge.loc.Length = uint32(size) + curMerge.blobs = append(curMerge.blobs, l) + continue + } + } + + rr = &rangeRead{ + loc: l.Location, + blobs: []blobLocation{l}, + } + plan = append(plan, rr) + } + + return plan +} diff --git a/repository/getblobs_internal_test.go b/repository/getblobs_internal_test.go new file mode 100644 index 00000000..fc5aeeb2 --- /dev/null +++ b/repository/getblobs_internal_test.go @@ -0,0 +1,290 @@ +package repository + +import ( + "math/rand/v2" + "testing" + + "github.com/PlakarKorp/kloset/objects" + "github.com/PlakarKorp/kloset/repository/state" + "github.com/PlakarKorp/kloset/resources" + "github.com/stretchr/testify/require" +) + +func bl(pf byte, offset uint64, length uint32) blobLocation { + return blobLocation{ + BlobReq{Type: resources.RT_CHUNK, MAC: objects.MAC{pf, byte(offset), byte(offset >> 8)}}, + state.Location{Packfile: objects.MAC{pf}, Offset: offset, Length: length}, + } +} + +func TestPlanReadsAdjacentMerge(t *testing.T) { + // Three contiguous blobs coalesce into a single zero-waste read. + plans := readPlanner([]blobLocation{ + bl(1, 200, 100), + bl(1, 0, 100), + bl(1, 100, 100), + }, maxSize, 0.3) + + require.Len(t, plans, 1) + require.Equal(t, uint64(0), plans[0].loc.Offset) + require.Equal(t, uint32(300), plans[0].loc.Length) + require.Len(t, plans[0].blobs, 3) +} + +func TestPlanReadsPackfileBoundary(t *testing.T) { + // Contiguous offsets in different packfiles never merge. + plans := readPlanner([]blobLocation{ + bl(1, 0, 100), + bl(2, 100, 100), + }, maxSize, 0.3) + + require.Len(t, plans, 2) +} + +func TestPlanReadsMaxReadSize(t *testing.T) { + // Merging stops when the merged read would exceed maxReadSize, and + // a single blob larger than maxReadSize still gets its own read. + plans := readPlanner([]blobLocation{ + bl(1, 0, 600), + bl(1, 600, 600), + bl(1, 1200, 2000), + }, 1000, 0.3) + + require.Len(t, plans, 3) + require.Equal(t, uint32(2000), plans[2].loc.Length) +} + +// checkPlan asserts the invariants any correct plan must satisfy, whatever +// the merge policy in force: +// +// 1. every input blob lands in exactly one plan (none dropped, none doubled); +// 2. a plan only contains blobs of its own packfile; +// 3. a plan's range covers every blob assigned to it, entirely; +// 4. a plan's range is tight: it starts at its first blob and ends at the +// end of its outermost blob — gaps between blobs are fetched, slack +// beyond them is not; +// 5. a multi-blob plan never exceeds maxReadSize (a single blob larger +// than maxReadSize is allowed a plan of its own). +func checkPlan(t *testing.T, input []blobLocation, plans []*rangeRead, maxReadSize uint32) { + t.Helper() + + seen := 0 + for i, p := range plans { + require.NotEmpty(t, p.blobs, "plan %d contains no blobs", i) + + planEnd := p.loc.Offset + uint64(p.loc.Length) + minOff, maxEnd := p.blobs[0].Offset, uint64(0) + for _, b := range p.blobs { + blobEnd := b.Offset + uint64(b.Length) + require.Equal(t, p.loc.Packfile, b.Packfile, + "plan %d: blob %x belongs to another packfile", i, b.MAC) + require.GreaterOrEqual(t, b.Offset, p.loc.Offset, + "plan %d: blob %x starts before the plan's range", i, b.MAC) + require.LessOrEqual(t, blobEnd, planEnd, + "plan %d: blob %x overruns the plan's range", i, b.MAC) + minOff = min(minOff, b.Offset) + maxEnd = max(maxEnd, blobEnd) + seen++ + } + + require.Equal(t, minOff, p.loc.Offset, + "plan %d fetches slack before its first blob", i) + require.Equal(t, maxEnd, planEnd, + "plan %d fetches slack past its outermost blob", i) + + if len(p.blobs) > 1 { + require.LessOrEqual(t, p.loc.Length, maxReadSize, + "plan %d: merged read exceeds maxReadSize", i) + } + } + require.Equal(t, len(input), seen, + "every input blob must appear in exactly one plan") +} + +// checkMaximal asserts that no two consecutive same-packfile plans could +// have been merged: their combined span must exceed maxReadSize. Unlike +// checkPlan this is specific to the everything-permitted policy +// (maxWaste 1.0) — with a real waste policy a split can be legitimate +// below the size cap. It catches the failure mode checkPlan cannot: +// a planner drifting back toward one read per blob would produce +// perfectly valid, tiny plans, and defeat the whole point. +// +// It assumes plans come out sorted by (packfile, offset), which the +// sequential greedy emission over sorted input guarantees. +func checkMaximal(t *testing.T, plans []*rangeRead, maxReadSize uint32) { + t.Helper() + + for i := 1; i < len(plans); i++ { + prev, cur := plans[i-1], plans[i] + if prev.loc.Packfile != cur.loc.Packfile { + continue + } + span := cur.loc.Offset + uint64(cur.loc.Length) - prev.loc.Offset + require.Greater(t, span, uint64(maxReadSize), + "plans %d and %d span %d bytes together and should have been merged", + i-1, i, span) + } +} + +func TestPlanReadsGapCountsTowardMaxReadSize(t *testing.T) { + // Two 100-byte blobs 800 bytes apart: the merged READ would be + // 1000 bytes even though only 200 are wanted. With maxReadSize + // 500 they must not merge — the cap bounds the fetched span, + // gaps included, not the sum of blob lengths. + input := []blobLocation{ + bl(1, 0, 100), + bl(1, 900, 100), + } + plans := readPlanner(input, 500, 1.0) + + require.Len(t, plans, 2) + checkPlan(t, input, plans, 500) +} + +func TestPlanReadsGapExtendsLength(t *testing.T) { + // Two blobs with a 100-byte gap, everything permitted: one plan, + // whose length spans the gap (300), not the sum of blob lengths + // (200) — otherwise the read misses the second blob's tail. + input := []blobLocation{ + bl(1, 0, 100), + bl(1, 200, 100), + } + plans := readPlanner(input, maxSize, 1.0) + + require.Len(t, plans, 1) + require.Equal(t, uint64(0), plans[0].loc.Offset) + require.Equal(t, uint32(300), plans[0].loc.Length) + checkPlan(t, input, plans, maxSize) +} + +func TestPlanReadsMaxReadSizeExact(t *testing.T) { + // A merge landing exactly on maxReadSize is allowed: the cap is + // inclusive ("reads up to maxReadSize"), not exclusive. + input := []blobLocation{ + bl(1, 0, 400), + bl(1, 600, 400), + } + plans := readPlanner(input, 1000, 1.0) + + require.Len(t, plans, 1) + require.Equal(t, uint32(1000), plans[0].loc.Length) + checkPlan(t, input, plans, 1000) +} + +func TestPlanReadsOversizedBlobDoesNotBlockNeighbours(t *testing.T) { + // A blob larger than maxReadSize gets a plan of its own, and the + // blobs after it still merge with each other. + input := []blobLocation{ + bl(1, 0, 2000), + bl(1, 2000, 100), + bl(1, 2100, 100), + } + plans := readPlanner(input, 1000, 1.0) + + require.Len(t, plans, 2) + require.Equal(t, uint32(2000), plans[0].loc.Length) + require.Equal(t, uint32(200), plans[1].loc.Length) + checkPlan(t, input, plans, 1000) +} + +func TestPlanReadsInterleavedPackfiles(t *testing.T) { + // Unsorted input alternating between two packfiles: blobs regroup + // per packfile and merge within it. + input := []blobLocation{ + bl(2, 100, 100), + bl(1, 100, 100), + bl(2, 0, 100), + bl(1, 0, 100), + } + plans := readPlanner(input, maxSize, 1.0) + + require.Len(t, plans, 2) + for _, p := range plans { + require.Equal(t, uint64(0), p.loc.Offset) + require.Equal(t, uint32(200), p.loc.Length) + } + checkPlan(t, input, plans, maxSize) +} + +func TestPlanReadsSingle(t *testing.T) { + input := []blobLocation{bl(1, 42, 100)} + plans := readPlanner(input, maxSize, 1.0) + + require.Len(t, plans, 1) + require.Equal(t, uint64(42), plans[0].loc.Offset) + require.Equal(t, uint32(100), plans[0].loc.Length) + checkPlan(t, input, plans, maxSize) +} + +func TestPlanReadsStress(t *testing.T) { + // Deterministic pseudo-random batches laid out the way blobs + // really sit in packfiles — disjoint ranges, sometimes adjacent, + // sometimes gapped, occasionally oversized — checked purely + // against the checkPlan invariants. Policy-agnostic on purpose: + // this must keep passing unchanged once the waste policy lands. + rng := rand.New(rand.NewPCG(0x9E3779B9, 0x7F4A7C15)) + + for round := range 20 { + maxReadSize := uint32(1024 + rng.IntN(8192)) + + input := make([]blobLocation, 0, 512) + for pf := byte(1); pf <= 3; pf++ { + var offset uint64 + for range 150 + rng.IntN(50) { + if rng.IntN(3) > 0 { // 1 in 3 blobs is adjacent to the previous one + offset += uint64(rng.IntN(3000)) + } + length := uint32(1 + rng.IntN(2000)) + if rng.IntN(50) == 0 { // occasional oversized blob + length = maxReadSize + uint32(rng.IntN(1000)) + } + input = append(input, bl(pf, offset, length)) + offset += uint64(length) + } + } + rng.Shuffle(len(input), func(i, j int) { + input[i], input[j] = input[j], input[i] + }) + + plans := readPlanner(input, maxReadSize, 1.0) + if t.Failed() { + return + } + t.Logf("round %d: maxReadSize=%d, %d blobs -> %d plans", + round, maxReadSize, len(input), len(plans)) + checkPlan(t, input, plans, maxReadSize) + checkMaximal(t, plans, maxReadSize) + } +} + +// BenchmarkReadPlanner runs at the scale of the measured no-change +// kernel-backup log (~162k blobs over 20 packfiles): planning must stay +// trivial next to a single ~80ms store round trip. +func BenchmarkReadPlanner(b *testing.B) { + rng := rand.New(rand.NewPCG(0xDEADBEEF, 0xCAFEBABE)) + + input := make([]blobLocation, 0, 162_000) + for pf := byte(1); pf <= 20; pf++ { + var offset uint64 + for range 8_100 { + if rng.IntN(3) > 0 { + offset += uint64(rng.IntN(3000)) + } + length := uint32(1 + rng.IntN(2000)) + input = append(input, bl(pf, offset, length)) + offset += uint64(length) + } + } + rng.Shuffle(len(input), func(i, j int) { + input[i], input[j] = input[j], input[i] + }) + + scratch := make([]blobLocation, len(input)) + b.ReportAllocs() + for b.Loop() { + // readPlanner sorts its input in place; replanning already + // sorted input would flatter the numbers. + copy(scratch, input) + readPlanner(scratch, maxSize, 1.0) + } +} diff --git a/repository/getblobs_test.go b/repository/getblobs_test.go new file mode 100644 index 00000000..98b28c22 --- /dev/null +++ b/repository/getblobs_test.go @@ -0,0 +1,139 @@ +package repository_test + +import ( + "context" + "fmt" + "strings" + "testing" + + "github.com/PlakarKorp/kloset/objects" + "github.com/PlakarKorp/kloset/repository" + "github.com/PlakarKorp/kloset/resources" + ptesting "github.com/PlakarKorp/kloset/testing" + "github.com/stretchr/testify/require" +) + +// chunkRequests backs up a few files and returns a BlobReq per chunk +// of every resolved object, so GetBlobs can be exercised against blobs +// that really live in packfiles. +func chunkRequests(t *testing.T, repo *repository.Repository, paths ...string) []repository.BlobReq { + files := []ptesting.MockFile{ptesting.NewMockDir("/")} + for i, p := range paths { + files = append(files, ptesting.NewMockFile(p, 0644, + strings.Repeat(fmt.Sprintf("get-blobs payload %d ", i), 64))) + } + snap := ptesting.GenerateSnapshot(t, repo, files) + require.NotNil(t, snap) + + fs, err := snap.Filesystem() + require.NoError(t, err) + + var reqs []repository.BlobReq + for _, p := range paths { + entry, err := fs.GetEntry(p) + require.NoError(t, err) + require.NotNil(t, entry.ResolvedObject) + for _, c := range entry.ResolvedObject.Chunks { + reqs = append(reqs, repository.BlobReq{Type: resources.RT_CHUNK, MAC: c.ContentMAC}) + } + } + require.NotEmpty(t, reqs) + return reqs +} + +func TestGetBlobsRoundtrip(t *testing.T) { + repo := ptesting.GenerateRepository(t, nil, nil, nil) + reqs := chunkRequests(t, repo, "/a.txt", "/b.txt", "/c.txt") + + // Duplicate every request: set semantics must collapse them. + got := make(map[repository.BlobReq][]byte) + for res, err := range repo.GetBlobs(context.Background(), append(reqs, reqs...), nil) { + require.NoError(t, err) + req := repository.BlobReq{Type: res.Type, MAC: res.MAC} + _, dup := got[req] + require.False(t, dup, "blob %x delivered twice", res.MAC) + got[req] = res.Data + } + + distinct := make(map[repository.BlobReq]struct{}) + for _, req := range reqs { + distinct[req] = struct{}{} + } + require.Len(t, got, len(distinct)) + + // Every blob must byte-match the single-blob read path. + for req, data := range got { + expected, err := repo.GetBlobBytes(req.Type, req.MAC) + require.NoError(t, err) + require.Equal(t, expected, data) + } +} + +func TestGetBlobsMissingBlob(t *testing.T) { + repo := ptesting.GenerateRepository(t, nil, nil, nil) + reqs := chunkRequests(t, repo, "/a.txt") + + bogus := objects.MAC{0xDE, 0xAD, 0xBE, 0xEF} + reqs = append(reqs, repository.BlobReq{Type: resources.RT_CHUNK, MAC: bogus}) + + var missing, found int + for res, err := range repo.GetBlobs(context.Background(), reqs, nil) { + if res.MAC == bogus { + require.ErrorIs(t, err, repository.ErrBlobNotFound) + require.Nil(t, res.Data) + missing++ + } else { + require.NoError(t, err) + found++ + } + } + require.Equal(t, 1, missing) + require.Equal(t, len(reqs)-1, found) +} + +func TestGetBlobsAllMissing(t *testing.T) { + // A batch where NO request resolves: every result carries + // ErrBlobNotFound and the iterator terminates cleanly. + repo := ptesting.GenerateRepository(t, nil, nil, nil) + _ = chunkRequests(t, repo, "/a.txt") // populate the repo, request none of it + + reqs := []repository.BlobReq{ + {Type: resources.RT_CHUNK, MAC: objects.MAC{0xDE, 0xAD}}, + {Type: resources.RT_CHUNK, MAC: objects.MAC{0xBE, 0xEF}}, + } + + count := 0 + for res, err := range repo.GetBlobs(context.Background(), reqs, nil) { + require.ErrorIs(t, err, repository.ErrBlobNotFound) + require.Nil(t, res.Data) + count++ + } + require.Equal(t, len(reqs), count) +} + +func TestGetBlobsRangeReadError(t *testing.T) { + // Blobs resolve in the state but the packfile is gone from the + // store: every blob of the failed range yields its identity plus + // the read error — no panic, no phantom success. + repo := ptesting.GenerateRepository(t, nil, nil, nil) + reqs := chunkRequests(t, repo, "/a.txt", "/b.txt") + + for mac := range repo.ListPackfiles() { + require.NoError(t, repo.DeletePackfile(mac)) + } + + count := 0 + for res, err := range repo.GetBlobs(context.Background(), reqs, nil) { + require.Error(t, err) + require.Nil(t, res.Data) + count++ + } + require.Equal(t, len(reqs), count) +} + +func TestGetBlobsEmpty(t *testing.T) { + repo := ptesting.GenerateRepository(t, nil, nil, nil) + for res, err := range repo.GetBlobs(context.Background(), nil, nil) { + t.Fatalf("unexpected result %x (err %v)", res.MAC, err) + } +}