mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-08-20 22:27:04 +00:00
fix(mount): count manifest sizes in merge condition to prevent accumulation (#9101)
* fix(mount): count manifest sizes in merge condition to prevent manifest accumulation shouldMergeChunks only counted non-manifest chunk sizes toward the bloat threshold, so overlapping manifests accumulated undetected across flush cycles. During sustained random writes (e.g. fio), each metadata flush compacts non-manifest chunks and may group them into a new manifest via MaybeManifestize, while carrying forward all existing manifests. Since the merge condition only checked non-manifest totals against 2x file size, the growing pile of redundant manifests never triggered a merge. In a real workload: 4 GB file, 25 manifests each covering ~4 GB (107 GB manifest data on volume servers), but shouldMergeChunks saw only 4.2 GB of non-manifest chunks vs the 8.6 GB threshold -- no merge. Fix: include manifest coverage sizes in totalChunkSize. This correctly detects the 111 GB total vs 8.6 GB threshold and triggers the merge, which re-reads the file as clean chunks and lets the filer's MinusChunks (which resolves manifests) garbage-collect all redundant sub-chunks. * test(mount,filer): add manifest chunk correctness tests Filer-level tests (filechunk_manifest_test.go): - TestManifestRoundTripPreservesChunks: create -> serialize -> deserialize preserves fileId, offset, size, and timestamp for all sub-chunks - TestCompactResolvedOverlappingManifests: two manifests covering the same range, older sub-chunks correctly identified as garbage after compaction - TestDoMinusChunksWithResolvedManifests: old manifest sub-chunks detected as garbage when compared against new clean chunks (mirrors filer cleanup) - TestManifestizeSmallBatchWithRemainder: batch=3 with 7 chunks produces 2 manifests + 1 remainder, all data recoverable after resolve - TestCompactMultipleOverlappingManifestGenerations: 5 generations of full-file manifests, compaction keeps only the newest generation - TestManifestBloatDetection: with N overlapping manifests, merge triggers at N >= 2 (stored > 2x file size) Mount-level test (weedfs_file_sync_test.go): - TestFlushCycleManifestAccumulation: simulates the flushMetadataToFiler pipeline over multiple cycles with a small manifest batch (5 chunks). Verifies merge triggers at cycle 2 when accumulated manifests exceed the 2x threshold. Uses SeparateManifestChunks + CompactFileChunks + shouldMergeChunks to mirror the real code path.
This commit is contained in:
Chris Lu
@@ -2,10 +2,14 @@ package filer
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"math"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"google.golang.org/protobuf/proto"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
)
|
||||
@@ -111,3 +115,392 @@ func mockMerge(saveFunc SaveDataAsChunkFunctionType, dataChunks []*filer_pb.File
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// In-memory manifest store for round-trip tests
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
// testManifestStore provides in-memory save/resolve for manifest tests
|
||||
// without requiring volume servers.
|
||||
type testManifestStore struct {
|
||||
stored map[string][]byte
|
||||
nextID int
|
||||
}
|
||||
|
||||
func newTestManifestStore() *testManifestStore {
|
||||
return &testManifestStore{stored: make(map[string][]byte)}
|
||||
}
|
||||
|
||||
func (s *testManifestStore) saveFunc() SaveDataAsChunkFunctionType {
|
||||
return func(reader io.Reader, name string, offset int64, tsNs int64) (*filer_pb.FileChunk, error) {
|
||||
data, err := io.ReadAll(reader)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
s.nextID++
|
||||
chunk := &filer_pb.FileChunk{
|
||||
Fid: &filer_pb.FileId{VolumeId: 999, FileKey: uint64(s.nextID), Cookie: 0},
|
||||
}
|
||||
s.stored[chunk.GetFileIdString()] = data
|
||||
return chunk, nil
|
||||
}
|
||||
}
|
||||
|
||||
// resolve expands a single manifest chunk by deserializing the stored proto.
|
||||
func (s *testManifestStore) resolve(chunk *filer_pb.FileChunk) ([]*filer_pb.FileChunk, error) {
|
||||
data, ok := s.stored[chunk.GetFileIdString()]
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("manifest %q not in store", chunk.GetFileIdString())
|
||||
}
|
||||
m := &filer_pb.FileChunkManifest{}
|
||||
if err := proto.Unmarshal(data, m); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
filer_pb.AfterEntryDeserialization(m.Chunks)
|
||||
return m.Chunks, nil
|
||||
}
|
||||
|
||||
// resolveAll expands all manifest chunks, returns (dataChunks, manifestChunks).
|
||||
func (s *testManifestStore) resolveAll(chunks []*filer_pb.FileChunk) (data, manifests []*filer_pb.FileChunk, err error) {
|
||||
for _, c := range chunks {
|
||||
if !c.IsChunkManifest {
|
||||
data = append(data, c)
|
||||
continue
|
||||
}
|
||||
manifests = append(manifests, c)
|
||||
sub, resolveErr := s.resolve(c)
|
||||
if resolveErr != nil {
|
||||
return nil, nil, resolveErr
|
||||
}
|
||||
data = append(data, sub...)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func testChunk(volId uint32, key uint64, cookie uint32, offset int64, size uint64, tsNs int64) *filer_pb.FileChunk {
|
||||
return &filer_pb.FileChunk{
|
||||
Fid: &filer_pb.FileId{VolumeId: volId, FileKey: key, Cookie: cookie},
|
||||
Offset: offset,
|
||||
Size: size,
|
||||
ModifiedTsNs: tsNs,
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Manifest round-trip: create -> serialize -> deserialize -> verify
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
func TestManifestRoundTripPreservesChunks(t *testing.T) {
|
||||
store := newTestManifestStore()
|
||||
|
||||
chunks := []*filer_pb.FileChunk{
|
||||
testChunk(1, 1, 100, 0, 1000, 10),
|
||||
testChunk(1, 2, 200, 1000, 1000, 20),
|
||||
testChunk(2, 3, 300, 2000, 1000, 30),
|
||||
}
|
||||
|
||||
origIds := make([]string, len(chunks))
|
||||
for i, c := range chunks {
|
||||
origIds[i] = c.GetFileIdString()
|
||||
}
|
||||
|
||||
// Use real mergeIntoManifest with batch=3 so all go into one manifest
|
||||
result, err := doMaybeManifestize(store.saveFunc(), chunks, 3, mergeIntoManifest)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(result) != 1 || !result[0].IsChunkManifest {
|
||||
t.Fatalf("expected 1 manifest chunk, got %d chunks", len(result))
|
||||
}
|
||||
if result[0].Offset != 0 || result[0].Size != 3000 {
|
||||
t.Errorf("manifest coverage: offset=%d size=%d, want 0/3000", result[0].Offset, result[0].Size)
|
||||
}
|
||||
|
||||
resolved, err := store.resolve(result[0])
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(resolved) != 3 {
|
||||
t.Fatalf("expected 3 sub-chunks, got %d", len(resolved))
|
||||
}
|
||||
for i, got := range resolved {
|
||||
if got.GetFileIdString() != origIds[i] {
|
||||
t.Errorf("chunk %d: fileId %q != %q", i, got.GetFileIdString(), origIds[i])
|
||||
}
|
||||
if got.Offset != chunks[i].Offset {
|
||||
t.Errorf("chunk %d: offset %d != %d", i, got.Offset, chunks[i].Offset)
|
||||
}
|
||||
if got.Size != chunks[i].Size {
|
||||
t.Errorf("chunk %d: size %d != %d", i, got.Size, chunks[i].Size)
|
||||
}
|
||||
if got.ModifiedTsNs != chunks[i].ModifiedTsNs {
|
||||
t.Errorf("chunk %d: ts %d != %d", i, got.ModifiedTsNs, chunks[i].ModifiedTsNs)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Compact resolved overlapping manifests: older sub-chunks become garbage
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
func TestCompactResolvedOverlappingManifests(t *testing.T) {
|
||||
store := newTestManifestStore()
|
||||
|
||||
// Batch 1: older writes covering [0, 3000)
|
||||
batch1 := []*filer_pb.FileChunk{
|
||||
testChunk(1, 1, 1, 0, 1000, 1),
|
||||
testChunk(1, 2, 2, 1000, 1000, 2),
|
||||
testChunk(1, 3, 3, 2000, 1000, 3),
|
||||
}
|
||||
origBatch1Ids := make(map[string]bool)
|
||||
for _, c := range batch1 {
|
||||
origBatch1Ids[c.GetFileIdString()] = true
|
||||
}
|
||||
|
||||
// Batch 2: newer writes covering [0, 3000)
|
||||
batch2 := []*filer_pb.FileChunk{
|
||||
testChunk(2, 1, 1, 0, 1000, 10),
|
||||
testChunk(2, 2, 2, 1000, 1000, 11),
|
||||
testChunk(2, 3, 3, 2000, 1000, 12),
|
||||
}
|
||||
origBatch2Ids := make(map[string]bool)
|
||||
for _, c := range batch2 {
|
||||
origBatch2Ids[c.GetFileIdString()] = true
|
||||
}
|
||||
|
||||
// Manifestize each batch separately (simulates successive flush cycles)
|
||||
save := store.saveFunc()
|
||||
manifest1, err := doMaybeManifestize(save, batch1, 3, mergeIntoManifest)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
manifest2, err := doMaybeManifestize(save, batch2, 3, mergeIntoManifest)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Resolve all manifests into sub-chunks
|
||||
allChunks := append(manifest1, manifest2...)
|
||||
dataChunks, _, err := store.resolveAll(allChunks)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(dataChunks) != 6 {
|
||||
t.Fatalf("expected 6 resolved data chunks, got %d", len(dataChunks))
|
||||
}
|
||||
|
||||
// CompactFileChunks on resolved sub-chunks (nil lookupFn is fine for non-manifest chunks)
|
||||
compacted, garbage := CompactFileChunks(context.Background(), nil, dataChunks)
|
||||
|
||||
if len(compacted) != 3 {
|
||||
t.Fatalf("expected 3 compacted chunks, got %d", len(compacted))
|
||||
}
|
||||
if len(garbage) != 3 {
|
||||
t.Fatalf("expected 3 garbage chunks, got %d", len(garbage))
|
||||
}
|
||||
|
||||
for _, c := range compacted {
|
||||
if !origBatch2Ids[c.GetFileIdString()] {
|
||||
t.Errorf("compacted chunk %q should be from newer batch", c.GetFileIdString())
|
||||
}
|
||||
}
|
||||
for _, c := range garbage {
|
||||
if !origBatch1Ids[c.GetFileIdString()] {
|
||||
t.Errorf("garbage chunk %q should be from older batch", c.GetFileIdString())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// DoMinusChunks: old manifest sub-chunks identified as garbage vs new chunks
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
func TestDoMinusChunksWithResolvedManifests(t *testing.T) {
|
||||
store := newTestManifestStore()
|
||||
|
||||
// Old entry: data packed into a manifest
|
||||
oldData := []*filer_pb.FileChunk{
|
||||
testChunk(1, 1, 1, 0, 1000, 1),
|
||||
testChunk(1, 2, 2, 1000, 1000, 2),
|
||||
testChunk(1, 3, 3, 2000, 1000, 3),
|
||||
}
|
||||
oldManifest, err := doMaybeManifestize(store.saveFunc(), oldData, 3, mergeIntoManifest)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Resolve old manifests into sub-chunks (what MinusChunks does internally)
|
||||
oldResolved, oldMeta, err := store.resolveAll(oldManifest)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// New entry: clean re-uploaded chunks (from merge) with different file IDs
|
||||
newData := []*filer_pb.FileChunk{
|
||||
testChunk(3, 10, 10, 0, 1500, 100),
|
||||
testChunk(3, 11, 11, 1500, 1500, 101),
|
||||
}
|
||||
|
||||
// DoMinusChunks: data sub-chunks in old but not in new
|
||||
deltaData := DoMinusChunks(oldResolved, newData)
|
||||
if len(deltaData) != 3 {
|
||||
t.Fatalf("expected 3 old data chunks as garbage, got %d", len(deltaData))
|
||||
}
|
||||
|
||||
// DoMinusChunks: manifest chunks in old but not in new
|
||||
deltaMeta := DoMinusChunks(oldMeta, nil)
|
||||
if len(deltaMeta) != 1 {
|
||||
t.Fatalf("expected 1 old manifest chunk as garbage, got %d", len(deltaMeta))
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Small-batch manifestize: verify sub-chunks and remainder handling
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
func TestManifestizeSmallBatchWithRemainder(t *testing.T) {
|
||||
store := newTestManifestStore()
|
||||
|
||||
// 7 chunks with batch=3: should produce 2 manifests + 1 remainder
|
||||
chunks := make([]*filer_pb.FileChunk, 7)
|
||||
for i := range chunks {
|
||||
chunks[i] = testChunk(1, uint64(i+1), uint32(i+1), int64(i*100), 100, int64(i+1))
|
||||
}
|
||||
|
||||
result, err := doMaybeManifestize(store.saveFunc(), chunks, 3, mergeIntoManifest)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
manifests := 0
|
||||
regular := 0
|
||||
for _, c := range result {
|
||||
if c.IsChunkManifest {
|
||||
manifests++
|
||||
} else {
|
||||
regular++
|
||||
}
|
||||
}
|
||||
if manifests != 2 {
|
||||
t.Errorf("expected 2 manifests, got %d", manifests)
|
||||
}
|
||||
if regular != 1 {
|
||||
t.Errorf("expected 1 remainder chunk, got %d", regular)
|
||||
}
|
||||
|
||||
// Resolve all and verify we get back 7 data chunks
|
||||
data, _, err := store.resolveAll(result)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(data) != 7 {
|
||||
t.Fatalf("expected 7 data chunks after resolve, got %d", len(data))
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Multiple overlapping manifests: compaction identifies all redundant sub-chunks
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
func TestCompactMultipleOverlappingManifestGenerations(t *testing.T) {
|
||||
store := newTestManifestStore()
|
||||
save := store.saveFunc()
|
||||
|
||||
const fileSize int64 = 3000
|
||||
const chunkSize uint64 = 1000
|
||||
|
||||
// Simulate 5 generations of full-file writes, each manifestized separately.
|
||||
var allManifests []*filer_pb.FileChunk
|
||||
for gen := 0; gen < 5; gen++ {
|
||||
tsBase := int64((gen + 1) * 100)
|
||||
batch := []*filer_pb.FileChunk{
|
||||
testChunk(uint32(gen+1), 1, uint32(gen), 0, chunkSize, tsBase),
|
||||
testChunk(uint32(gen+1), 2, uint32(gen), int64(chunkSize), chunkSize, tsBase+1),
|
||||
testChunk(uint32(gen+1), 3, uint32(gen), int64(chunkSize*2), chunkSize, tsBase+2),
|
||||
}
|
||||
manifest, err := doMaybeManifestize(save, batch, 3, mergeIntoManifest)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
allManifests = append(allManifests, manifest...)
|
||||
}
|
||||
|
||||
if len(allManifests) != 5 {
|
||||
t.Fatalf("expected 5 manifest chunks, got %d", len(allManifests))
|
||||
}
|
||||
|
||||
// Resolve all 5 manifests: 15 data chunks total
|
||||
dataChunks, _, err := store.resolveAll(allManifests)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(dataChunks) != 15 {
|
||||
t.Fatalf("expected 15 resolved chunks, got %d", len(dataChunks))
|
||||
}
|
||||
|
||||
// Compact: only generation 5 (the newest 3 chunks) should survive
|
||||
compacted, garbage := CompactFileChunks(context.Background(), nil, dataChunks)
|
||||
if len(compacted) != 3 {
|
||||
t.Errorf("expected 3 compacted (gen 5), got %d", len(compacted))
|
||||
}
|
||||
if len(garbage) != 12 {
|
||||
t.Errorf("expected 12 garbage (gens 1-4), got %d", len(garbage))
|
||||
}
|
||||
|
||||
// Verify the surviving chunks are the newest (highest timestamps)
|
||||
for _, c := range compacted {
|
||||
if c.ModifiedTsNs < 500 {
|
||||
t.Errorf("compacted chunk ts=%d should be from gen 5 (ts >= 500)", c.ModifiedTsNs)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Bloat detection: manifest sizes must count toward total
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
func TestManifestBloatDetection(t *testing.T) {
|
||||
// Simulates what shouldMergeChunks sees after multiple flush cycles.
|
||||
// With the fix, manifest sizes count toward totalChunkSize, so bloat
|
||||
// from accumulated manifests is detected.
|
||||
|
||||
const fileSize uint64 = 10000
|
||||
|
||||
// 1 regular chunk covering the file (from latest compaction)
|
||||
regular := []*filer_pb.FileChunk{
|
||||
{Offset: 0, Size: fileSize, FileId: "latest", ModifiedTsNs: 100,
|
||||
Fid: &filer_pb.FileId{VolumeId: 1, FileKey: 1}},
|
||||
}
|
||||
|
||||
// N manifests each covering the full file (from prior flush cycles)
|
||||
for n := 1; n <= 10; n++ {
|
||||
var manifests []*filer_pb.FileChunk
|
||||
for i := 0; i < n; i++ {
|
||||
manifests = append(manifests, &filer_pb.FileChunk{
|
||||
Offset: 0, Size: fileSize, IsChunkManifest: true,
|
||||
Fid: &filer_pb.FileId{VolumeId: 900, FileKey: uint64(i + 1)},
|
||||
})
|
||||
}
|
||||
|
||||
allChunks := append(append([]*filer_pb.FileChunk{}, regular...), manifests...)
|
||||
var totalStored uint64
|
||||
for _, c := range allChunks {
|
||||
totalStored += c.Size
|
||||
}
|
||||
totalFile := TotalSize(allChunks)
|
||||
|
||||
expectMerge := totalStored > 2*totalFile
|
||||
t.Logf("n=%d manifests: totalStored=%d fileSize=%d ratio=%.1fx expectMerge=%v",
|
||||
n, totalStored, totalFile, float64(totalStored)/float64(totalFile), expectMerge)
|
||||
|
||||
// With 1 regular (10000) + N manifests (N*10000):
|
||||
// total = (N+1)*10000, fileSize=10000, ratio = N+1
|
||||
// Merge when N+1 > 2, i.e., N >= 2
|
||||
if n >= 2 && !expectMerge {
|
||||
t.Errorf("n=%d: should trigger merge at ratio %.1f", n, float64(totalStored)/float64(totalFile))
|
||||
}
|
||||
if n < 2 && expectMerge {
|
||||
t.Errorf("n=%d: should NOT trigger merge at ratio %.1f", n, float64(totalStored)/float64(totalFile))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -268,6 +268,13 @@ func shouldMergeChunks(compactedChunks []*filer_pb.FileChunk, manifestChunks []*
|
||||
for _, chunk := range compactedChunks {
|
||||
totalChunkSize += chunk.Size
|
||||
}
|
||||
// Count manifest coverage toward stored total. Each manifest holds
|
||||
// sub-chunks on volume servers that cover approximately Size bytes.
|
||||
// Without this, overlapping manifests accumulate undetected because
|
||||
// the merge condition only saw the (small) non-manifest chunk total.
|
||||
for _, chunk := range manifestChunks {
|
||||
totalChunkSize += chunk.Size
|
||||
}
|
||||
allChunks := make([]*filer_pb.FileChunk, 0, len(compactedChunks)+len(manifestChunks))
|
||||
allChunks = append(allChunks, compactedChunks...)
|
||||
allChunks = append(allChunks, manifestChunks...)
|
||||
|
||||
@@ -2,6 +2,7 @@ package mount
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"math"
|
||||
"math/rand/v2"
|
||||
"testing"
|
||||
@@ -68,7 +69,8 @@ func TestShouldMergeChunks_JustOverDouble(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestShouldMergeChunks_ManifestExtendFileSize(t *testing.T) {
|
||||
// Compacted chunks are small relative to file extended by manifest.
|
||||
// One manifest extends the file. Total stored = 100 (regular) + 900
|
||||
// (manifest) = 1000 vs 2*1000 = 2000 → no merge.
|
||||
compacted := []*filer_pb.FileChunk{
|
||||
{Offset: 0, Size: 100, FileId: "a", ModifiedTsNs: 1},
|
||||
}
|
||||
@@ -77,12 +79,13 @@ func TestShouldMergeChunks_ManifestExtendFileSize(t *testing.T) {
|
||||
}
|
||||
_, _, merge := shouldMergeChunks(compacted, manifest)
|
||||
if merge {
|
||||
t.Fatal("manifest-extended file should not merge when compacted chunks are small")
|
||||
t.Fatal("single manifest extending file should not merge")
|
||||
}
|
||||
}
|
||||
|
||||
func TestShouldMergeChunks_ManifestChunksSizeIgnored(t *testing.T) {
|
||||
// Only compacted chunk sizes count toward totalChunkSize.
|
||||
func TestShouldMergeChunks_ManifestSizesCounted(t *testing.T) {
|
||||
// Manifest sizes must count toward totalChunkSize so that overlapping
|
||||
// manifests trigger merge.
|
||||
compacted := []*filer_pb.FileChunk{
|
||||
{Offset: 0, Size: 100, FileId: "a", ModifiedTsNs: 1},
|
||||
}
|
||||
@@ -90,11 +93,35 @@ func TestShouldMergeChunks_ManifestChunksSizeIgnored(t *testing.T) {
|
||||
{Offset: 0, Size: 100, FileId: "m", ModifiedTsNs: 2, IsChunkManifest: true},
|
||||
}
|
||||
total, _, merge := shouldMergeChunks(compacted, manifest)
|
||||
if total != 100 {
|
||||
t.Fatalf("totalChunkSize should only count compacted, got %d", total)
|
||||
if total != 200 {
|
||||
t.Fatalf("totalChunkSize should count compacted + manifests, got %d", total)
|
||||
}
|
||||
if merge {
|
||||
t.Fatal("should not merge")
|
||||
t.Fatal("2x total on 100-byte file should not merge (need >2x)")
|
||||
}
|
||||
}
|
||||
|
||||
func TestShouldMergeChunks_AccumulatedManifests(t *testing.T) {
|
||||
// Simulates the real bug: multiple manifests each covering the full file
|
||||
// accumulate across flush cycles. Without counting manifest sizes, the
|
||||
// merge condition never fires and storage bloats indefinitely.
|
||||
compacted := []*filer_pb.FileChunk{
|
||||
{Offset: 0, Size: 1000, FileId: "a", ModifiedTsNs: 100},
|
||||
}
|
||||
// 5 manifests each covering the full file — successive flush cycles
|
||||
var manifests []*filer_pb.FileChunk
|
||||
for i := 0; i < 5; i++ {
|
||||
manifests = append(manifests, &filer_pb.FileChunk{
|
||||
Offset: 0, Size: 1000, FileId: string(rune('m' + i)),
|
||||
IsChunkManifest: true,
|
||||
})
|
||||
}
|
||||
total, fileSize, merge := shouldMergeChunks(compacted, manifests)
|
||||
// total = 1000 (regular) + 5*1000 (manifests) = 6000
|
||||
// fileSize = 1000
|
||||
// 6000 > 2*1000 → merge
|
||||
if !merge {
|
||||
t.Fatalf("accumulated overlapping manifests should trigger merge (total=%d fileSize=%d)", total, fileSize)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -279,6 +306,80 @@ func TestRandomWritesBloatDetection(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestFlushCycleManifestAccumulation simulates the flushMetadataToFiler
|
||||
// pipeline over multiple cycles with a small manifest batch threshold.
|
||||
// Each cycle adds a full pass of random writes, compacts, and manifestizes.
|
||||
// Verifies that shouldMergeChunks detects manifest bloat within a few cycles.
|
||||
func TestFlushCycleManifestAccumulation(t *testing.T) {
|
||||
const (
|
||||
fileSize int64 = 10000
|
||||
chunkSize uint64 = 1000
|
||||
numSlots = int(fileSize / int64(chunkSize)) // 10 offsets
|
||||
batchSize = 5 // small batch to trigger manifestize
|
||||
)
|
||||
|
||||
var entryChunks []*filer_pb.FileChunk
|
||||
nextTs := int64(1)
|
||||
nextKey := uint64(1)
|
||||
|
||||
for cycle := 0; cycle < 20; cycle++ {
|
||||
// Each cycle: new writes at every offset (simulates a full random-write pass)
|
||||
for slot := 0; slot < numSlots; slot++ {
|
||||
entryChunks = append(entryChunks, &filer_pb.FileChunk{
|
||||
Offset: int64(slot) * int64(chunkSize), Size: chunkSize,
|
||||
FileId: fmt.Sprintf("%d,%x00000000", cycle+1, nextKey),
|
||||
ModifiedTsNs: nextTs,
|
||||
Fid: &filer_pb.FileId{VolumeId: uint32(cycle + 1), FileKey: nextKey, Cookie: 0},
|
||||
})
|
||||
nextTs++
|
||||
nextKey++
|
||||
}
|
||||
|
||||
// --- flush pipeline (mirrors flushMetadataToFiler) ---
|
||||
manifestChunks, nonManifestChunks := filer.SeparateManifestChunks(entryChunks)
|
||||
compacted, _ := filer.CompactFileChunks(context.Background(), nil, nonManifestChunks)
|
||||
|
||||
_, _, merge := shouldMergeChunks(compacted, manifestChunks)
|
||||
if merge {
|
||||
t.Logf("cycle %d: merge correctly triggered (%d manifests, %d regular chunks)",
|
||||
cycle, len(manifestChunks), len(compacted))
|
||||
return
|
||||
}
|
||||
|
||||
// Simulate MaybeManifestize with small batch:
|
||||
// pack groups of batchSize regular chunks into manifest chunks.
|
||||
numPacked := len(compacted) / batchSize
|
||||
var newEntry []*filer_pb.FileChunk
|
||||
newEntry = append(newEntry, manifestChunks...) // carry forward old manifests
|
||||
|
||||
for i := 0; i < numPacked; i++ {
|
||||
batch := compacted[i*batchSize : (i+1)*batchSize]
|
||||
var minOff int64 = math.MaxInt64
|
||||
var maxEnd int64 = 0
|
||||
for _, c := range batch {
|
||||
if c.Offset < minOff {
|
||||
minOff = c.Offset
|
||||
}
|
||||
if end := c.Offset + int64(c.Size); end > maxEnd {
|
||||
maxEnd = end
|
||||
}
|
||||
}
|
||||
nextKey++
|
||||
newEntry = append(newEntry, &filer_pb.FileChunk{
|
||||
Offset: minOff, Size: uint64(maxEnd - minOff),
|
||||
FileId: fmt.Sprintf("900,%x00000000", nextKey),
|
||||
IsChunkManifest: true,
|
||||
Fid: &filer_pb.FileId{VolumeId: 900 + uint32(cycle), FileKey: nextKey, Cookie: 0},
|
||||
})
|
||||
}
|
||||
// Leftover regular chunks that didn't fill a batch
|
||||
remainder := compacted[numPacked*batchSize:]
|
||||
newEntry = append(newEntry, remainder...)
|
||||
entryChunks = newEntry
|
||||
}
|
||||
t.Fatal("merge never triggered after 20 flush cycles")
|
||||
}
|
||||
|
||||
// TestVisibleContentPreservedAfterCompact verifies that CompactFileChunks
|
||||
// never drops a chunk that contributes visible bytes. This is the invariant
|
||||
// that maybeMergeChunks relies on: the compacted set covers the same logical
|
||||
|
||||
Reference in New Issue
Block a user