Some checks failed
go: build dev binaries / cleanup (push) Successful in 4s
Codespell / Check for spelling errors (push) Failing after 3s
docker: build dev containers / build-rust-binaries (amd64, x86_64-unknown-linux-musl) (push) Failing after 3s
docker: build dev containers / build-dev-containers (push) Has been skipped
docker: build dev containers / build-rust-binaries (arm64, true, aarch64-unknown-linux-musl) (push) Failing after 3s
End to End / FUSE Mount (push) Failing after 2s
EC Integration Tests / EC Integration Tests (push) Failing after 30s
FUSE DLM Integration Tests / FUSE DLM Integration Tests (push) Failing after 3s
FUSE Volume Server Failover Tests / FUSE Volume Server Failover (push) Failing after 3s
FUSE Integration Tests / FUSE Integration Testing (push) Failing after 2s
FUSE P2P Peer Chunk Sharing Integration Tests / FUSE P2P Peer Chunk Sharing (push) Failing after 3s
go: build binary / Go Vet (push) Failing after 2s
go: build binary / Go Vet 32-bit (push) Failing after 3s
go: build binary / Build (push) Failing after 4s
go: build binary / Build cross-platform (freebsd/amd64) (push) Failing after 3s
go: build binary / Build cross-platform (windows/amd64) (push) Failing after 3s
go: build binary / Build cross-platform (darwin/arm64) (push) Failing after 3s
go: build binary / Build cross-platform (windows/arm64) (push) Failing after 3s
go: build binary / Test (push) Failing after 2s
go: build binary / Test 32-bit (push) Failing after 3s
helm: lint and test charts / lint-test (push) Failing after 3s
Java Client Integration Tests / Java Integration Tests (11) (push) Failing after 3s
Java Client Integration Tests / Java Integration Tests (17) (push) Failing after 3s
go: build dev binaries / build_dev_linux_windows (amd64, linux) (push) Failing after 2s
go: build dev binaries / build_dev_linux_windows (amd64, windows) (push) Failing after 3s
go: build dev binaries / build_dev_darwin (amd64, darwin) (push) Has been skipped
go: build dev binaries / build_dev_darwin (arm64, darwin) (push) Has been skipped
Java Client Unit Tests / Java Unit Tests (11) (push) Failing after 3s
Java Client Unit Tests / Java Unit Tests (17) (push) Failing after 2s
Java Client Unit Tests / Java Unit Tests (21) (push) Failing after 3s
Java Client Unit Tests / Java Unit Tests (8) (push) Failing after 3s
Kafka Quick Test (Load Test with Schema Registry) / Kafka Client Load Test (Quick) (push) Failing after 3s
Kafka Gateway Tests / Kafka Unit Tests (unit-tests-1) (push) Failing after 11s
Kafka Gateway Tests / Kafka Integration Tests (Critical) (integration-1) (push) Failing after 3s
Kafka Gateway Tests / Kafka End-to-End Tests (with SMQ) (e2e-1) (push) Failing after 3s
Kafka Gateway Tests / Kafka Consumer Group Tests (Highly Isolated) (consumer-group-1) (push) Failing after 2s
Kafka Gateway Tests / Kafka Client Compatibility (with SMQ) (client-compat-1) (push) Failing after 3s
Kafka Gateway Tests / Kafka SMQ Integration Tests (Full Stack) (smq-integration-1) (push) Failing after 3s
Kafka Gateway Tests / Kafka Protocol Tests (Isolated) (protocol-1) (push) Failing after 3s
KMS Tests / KMS Provider Integration Tests (push) Failing after 2s
KMS Tests / S3 KMS End-to-End Tests (push) Failing after 3s
Master Cold Start Tests / Master Cold Start Tests (push) Failing after 4s
Metadata Subscribe Integration Tests / Metadata Subscribe Integration Tests (push) Failing after 3s
Multi-Master Tests / Multi-Master Failover Tests (push) Failing after 3s
Performance / CPU and Heap Profile (push) Failing after 2s
Performance / S3 Read/Write Benchmark (rust volume) (push) Failing after 3s
pjdfstest POSIX Compliance / pjdfstest (push) Failing after 2s
Plugin Worker Integration Tests / Plugin Worker: erasure_coding (push) Failing after 3s
Plugin Worker Integration Tests / Plugin Worker: vacuum (push) Failing after 3s
Rust Volume Server Tests / Rust Integration Tests (push) Failing after 3s
Rust Volume Server Tests / Go Tests with Rust Volume (grpc - Shard 1) (push) Failing after 2s
Rust Volume Server Tests / Go Tests with Rust Volume (http - Shard 1) (push) Failing after 3s
Rust Volume Server Tests / Go Tests with Rust Volume (grpc - Shard 2) (push) Failing after 3s
Rust Volume Server Tests / Go Tests with Rust Volume (http - Shard 2) (push) Failing after 3s
Rust Volume Server Tests / Go Tests with Rust Volume (grpc - Shard 3) (push) Failing after 3s
S3 IAM Integration Tests / S3 IAM Distributed Tests (push) Failing after 2s
S3 IAM Integration Tests / S3 IAM Performance Tests (push) Failing after 3s
Performance / Throughput Benchmark (go volume) (push) Failing after 3s
Performance / Throughput Benchmark (rust volume) (push) Failing after 2s
Performance / S3 Read/Write Benchmark (go volume) (push) Failing after 3s
Plugin Worker Integration Tests / Plugin Worker: volume_balance (push) Failing after 3s
PostgreSQL Gateway Tests / PostgreSQL Basic Tests (push) Failing after 3s
Rust Volume Server Tests / Rust Unit Tests (push) Failing after 2s
Rust Volume Server Tests / Go Tests with Rust Volume (http - Shard 3) (push) Failing after 3s
Rust Plugin Worker Tests / Rust Plugin Worker Build and Unit Tests (push) Failing after 3s
rust: build dev volume server binaries / cleanup (push) Successful in 3s
S3 ETag and ACL Tests / S3 ETag + ACL Integration Tests (push) Failing after 3s
S3 IAM Integration Tests / IAM Unit Tests (push) Failing after 3s
S3 IAM Integration Tests / S3 IAM Integration Tests (advanced) (push) Failing after 3s
S3 IAM Integration Tests / S3 IAM Integration Tests (basic) (push) Failing after 2s
S3 IAM Integration Tests / S3 IAM Integration Tests (group) (push) Failing after 3s
S3 IAM Integration Tests / S3 IAM Integration Tests (policy-enforcement) (push) Failing after 2s
S3 IAM Integration Tests / S3 IAM Integration Tests (sts) (push) Failing after 4s
S3 Keycloak Integration Tests / S3 Keycloak Integration Tests (push) Failing after 3s
S3 PyArrow Parquet Tests / PyArrow Parquet Tests (Python 3.11) (push) Failing after 3s
S3 PyArrow Parquet Tests / Go Unit Tests (Implicit Directory) (push) Has been cancelled
S3 PyArrow Parquet Tests / PyArrow Parquet Tests (Python 3.12) (push) Has been cancelled
rust: build dev volume server binaries / build-rust-volume-dev-darwin (darwin-amd64, x86_64-apple-darwin) (push) Has been cancelled
rust: build dev volume server binaries / build-rust-volume-dev-darwin (darwin-arm64, aarch64-apple-darwin) (push) Has been cancelled
rust: build dev volume server binaries / build-rust-volume-dev-linux (linux-amd64, x86_64-unknown-linux-gnu) (push) Has been cancelled
S3 PyArrow Parquet Tests / PyArrow Parquet Tests (Python 3.9) (push) Has been cancelled
S3 Proxy Signature Tests / S3 Proxy Signature Verification Tests (push) Has been cancelled
S3 SDK V2 Route Disambiguation Tests / S3 SDK V2 Routing Tests (push) Has been cancelled
S3 SSE Tests / S3 SSE Integration Tests (quick) (push) Has been cancelled
S3 SSE Tests / S3 SSE Compatibility Test (push) Has been cancelled
S3 SSE Tests / S3 SSE Metadata Persistence Test (push) Has been cancelled
S3 SSE Tests / S3 SSE Copy Operations Test (push) Has been cancelled
S3 SSE Tests / S3 SSE Multipart Upload Test (push) Has been cancelled
S3 SSE Tests / S3 SSE Performance Test (push) Has been cancelled
S3 SSE Tests / S3 Volume Encryption Test (push) Has been cancelled
S3 SSE Tests / S3 SSE Integration Tests (comprehensive) (push) Has been cancelled
Spark Integration Tests / Spark Integration Tests (push) Failing after 2s
Telemetry Integration Tests / telemetry-integration-test (push) Failing after 3s
Volume Server Integration Tests / Volume Server Integration Tests (grpc - Shard 1) (push) Failing after 3s
Volume Server Integration Tests / Volume Server Integration Tests (http - Shard 1) (push) Failing after 3s
Ceph S3 tests / Basic S3 tests (KV store) (push) Has been cancelled
Ceph S3 tests / S3 Versioning & Object Lock tests (push) Has been cancelled
S3 Policy Integration Tests / S3 Policy Unit Tests (push) Failing after 3s
Ceph S3 tests / S3 CORS tests (push) Has been cancelled
Ceph S3 tests / SeaweedFS Custom S3 Copy tests (push) Has been cancelled
Ceph S3 tests / Basic S3 tests (SQL store) (push) Has been cancelled
test s3 over https using aws-cli / awscli-tests (push) Has been cancelled
TLS Rotation Integration Tests / TLS Rotation Integration Tests (push) Has been cancelled
Vacuum Integration Tests / Vacuum Integration Tests (push) Has been cancelled
S3 Policy Integration Tests / S3 Policy Variables Integration Tests (push) Failing after 3s
S3 Policy Integration Tests / S3 Policy Enforcement Integration Tests (advanced-policy) (push) Failing after 3s
S3 Policy Integration Tests / S3 Policy Enforcement Integration Tests (basic-policy) (push) Failing after 2s
S3 Policy Integration Tests / S3 Policy Enforcement Integration Tests (contextual-policy) (push) Failing after 4s
S3 Policy Integration Tests / Trusted Proxy Detection Tests (push) Failing after 2s
S3 Policy Integration Tests / S3 Policy Shell Integration Tests (push) Failing after 3s
Samba on FUSE Integration / samba-integration (push) Failing after 2s
SFTP Integration Tests / SFTP Integration Testing (push) Failing after 3s
terraform: validate and test modules / fmt, validate, plan-level tests (push) Failing after 3s
terraform: validate and test modules / local cluster smoke test (real weed) (push) Failing after 2s
Volume Server Integration Tests / Volume Server Integration Tests (grpc - Shard 3) (push) Failing after 3s
Volume Server Integration Tests / Volume Server Integration Tests (grpc - Shard 2) (push) Failing after 3s
Volume Server Integration Tests / Volume Server Integration Tests (http - Shard 2) (push) Failing after 2s
Volume Server Integration Tests / Volume Server Integration Tests (http - Shard 3) (push) Failing after 5s
mount: windows conformance / WinFsp conformance (push) Has been cancelled
mount: windows / Mount on Windows (push) Has been cancelled
445 lines
15 KiB
Go
445 lines
15 KiB
Go
package mount
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"math"
|
|
"math/rand/v2"
|
|
"testing"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/filer"
|
|
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
|
)
|
|
|
|
func TestShouldMergeChunks_NoChunks(t *testing.T) {
|
|
_, _, merge := shouldMergeChunks(nil, nil)
|
|
if merge {
|
|
t.Fatal("empty file should not trigger merge")
|
|
}
|
|
}
|
|
|
|
func TestShouldMergeChunks_SingleChunk(t *testing.T) {
|
|
chunks := []*filer_pb.FileChunk{
|
|
{Offset: 0, Size: 1000, FileId: "a", ModifiedTsNs: 1},
|
|
}
|
|
total, fileSize, merge := shouldMergeChunks(chunks, nil)
|
|
if merge {
|
|
t.Fatalf("single chunk should not merge (total=%d fileSize=%d)", total, fileSize)
|
|
}
|
|
}
|
|
|
|
func TestShouldMergeChunks_NonOverlapping(t *testing.T) {
|
|
chunks := []*filer_pb.FileChunk{
|
|
{Offset: 0, Size: 100, FileId: "a", ModifiedTsNs: 1},
|
|
{Offset: 100, Size: 100, FileId: "b", ModifiedTsNs: 2},
|
|
{Offset: 200, Size: 100, FileId: "c", ModifiedTsNs: 3},
|
|
}
|
|
total, fileSize, merge := shouldMergeChunks(chunks, nil)
|
|
if merge {
|
|
t.Fatalf("non-overlapping chunks should not merge (total=%d fileSize=%d)", total, fileSize)
|
|
}
|
|
if total != 300 || fileSize != 300 {
|
|
t.Fatalf("expected total=300 fileSize=300, got total=%d fileSize=%d", total, fileSize)
|
|
}
|
|
}
|
|
|
|
func TestShouldMergeChunks_ExactlyDouble(t *testing.T) {
|
|
// totalChunkSize == 2 * fileSize: should NOT merge (condition is >2x)
|
|
chunks := []*filer_pb.FileChunk{
|
|
{Offset: 0, Size: 100, FileId: "a", ModifiedTsNs: 1},
|
|
{Offset: 0, Size: 100, FileId: "b", ModifiedTsNs: 2},
|
|
}
|
|
_, _, merge := shouldMergeChunks(chunks, nil)
|
|
if merge {
|
|
t.Fatal("exactly 2x should not trigger merge")
|
|
}
|
|
}
|
|
|
|
func TestShouldMergeChunks_JustOverDouble(t *testing.T) {
|
|
// Three chunks overwriting the same range: 300 stored, 100 content = 3x
|
|
chunks := []*filer_pb.FileChunk{
|
|
{Offset: 0, Size: 100, FileId: "a", ModifiedTsNs: 1},
|
|
{Offset: 0, Size: 100, FileId: "b", ModifiedTsNs: 2},
|
|
{Offset: 0, Size: 100, FileId: "c", ModifiedTsNs: 3},
|
|
}
|
|
total, fileSize, merge := shouldMergeChunks(chunks, nil)
|
|
if !merge {
|
|
t.Fatalf("3x bloat should trigger merge (total=%d fileSize=%d)", total, fileSize)
|
|
}
|
|
}
|
|
|
|
func TestShouldMergeChunks_ManifestExtendFileSize(t *testing.T) {
|
|
// 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},
|
|
}
|
|
manifest := []*filer_pb.FileChunk{
|
|
{Offset: 100, Size: 900, FileId: "m", ModifiedTsNs: 2, IsChunkManifest: true},
|
|
}
|
|
_, _, merge := shouldMergeChunks(compacted, manifest)
|
|
if merge {
|
|
t.Fatal("single manifest extending file should not merge")
|
|
}
|
|
}
|
|
|
|
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},
|
|
}
|
|
manifest := []*filer_pb.FileChunk{
|
|
{Offset: 0, Size: 100, FileId: "m", ModifiedTsNs: 2, IsChunkManifest: true},
|
|
}
|
|
total, _, merge := shouldMergeChunks(compacted, manifest)
|
|
if total != 200 {
|
|
t.Fatalf("totalChunkSize should count compacted + manifests, got %d", total)
|
|
}
|
|
if 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)
|
|
}
|
|
}
|
|
|
|
func TestShouldMergeChunks_DoesNotMutateInput(t *testing.T) {
|
|
compacted := make([]*filer_pb.FileChunk, 2, 4) // extra capacity
|
|
compacted[0] = &filer_pb.FileChunk{Offset: 0, Size: 100, FileId: "a", ModifiedTsNs: 1}
|
|
compacted[1] = &filer_pb.FileChunk{Offset: 0, Size: 100, FileId: "b", ModifiedTsNs: 2}
|
|
manifest := []*filer_pb.FileChunk{
|
|
{Offset: 100, Size: 100, FileId: "m", ModifiedTsNs: 3, IsChunkManifest: true},
|
|
}
|
|
|
|
lenBefore := len(compacted)
|
|
shouldMergeChunks(compacted, manifest)
|
|
|
|
if len(compacted) != lenBefore {
|
|
t.Fatalf("shouldMergeChunks mutated compactedChunks: len changed from %d to %d",
|
|
lenBefore, len(compacted))
|
|
}
|
|
}
|
|
|
|
// TestCompactThenMergeCondition verifies that after CompactFileChunks
|
|
// removes fully-superseded chunks, the merge condition correctly detects
|
|
// remaining bloat from partially-overlapping chunks.
|
|
func TestCompactThenMergeCondition(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
chunks []*filer_pb.FileChunk
|
|
wantMerge bool
|
|
}{
|
|
{
|
|
name: "fully superseded chunks removed, no bloat",
|
|
chunks: []*filer_pb.FileChunk{
|
|
{Offset: 0, Size: 100, FileId: "old", ModifiedTsNs: 1, Fid: &filer_pb.FileId{FileKey: 1}},
|
|
{Offset: 0, Size: 100, FileId: "new", ModifiedTsNs: 2, Fid: &filer_pb.FileId{FileKey: 2}},
|
|
},
|
|
wantMerge: false,
|
|
},
|
|
{
|
|
name: "repeated full overwrites all compacted away",
|
|
chunks: []*filer_pb.FileChunk{
|
|
{Offset: 0, Size: 100, FileId: "a", ModifiedTsNs: 1, Fid: &filer_pb.FileId{FileKey: 1}},
|
|
{Offset: 0, Size: 100, FileId: "b", ModifiedTsNs: 2, Fid: &filer_pb.FileId{FileKey: 2}},
|
|
{Offset: 0, Size: 100, FileId: "c", ModifiedTsNs: 3, Fid: &filer_pb.FileId{FileKey: 3}},
|
|
{Offset: 0, Size: 100, FileId: "d", ModifiedTsNs: 4, Fid: &filer_pb.FileId{FileKey: 4}},
|
|
},
|
|
wantMerge: false,
|
|
},
|
|
{
|
|
name: "non-overlapping small chunks, no bloat",
|
|
chunks: func() []*filer_pb.FileChunk {
|
|
var c []*filer_pb.FileChunk
|
|
for i := 0; i < 10; i++ {
|
|
c = append(c, &filer_pb.FileChunk{
|
|
Offset: int64(i * 10),
|
|
Size: 10,
|
|
FileId: string(rune('a' + i)),
|
|
ModifiedTsNs: int64(i + 1),
|
|
Fid: &filer_pb.FileId{FileKey: uint64(i)},
|
|
})
|
|
}
|
|
return c
|
|
}(),
|
|
wantMerge: false,
|
|
},
|
|
{
|
|
name: "staggered partial overlaps create bloat",
|
|
// 100-byte base + three 40-byte writes at staggered offsets.
|
|
// All 4 survive compact (each partially visible).
|
|
// total = 100+40+40+40 = 220, file = 100, ratio = 2.2.
|
|
chunks: []*filer_pb.FileChunk{
|
|
{Offset: 0, Size: 100, FileId: "orig", ModifiedTsNs: 1, Fid: &filer_pb.FileId{FileKey: 1}},
|
|
{Offset: 10, Size: 40, FileId: "w1", ModifiedTsNs: 2, Fid: &filer_pb.FileId{FileKey: 2}},
|
|
{Offset: 30, Size: 40, FileId: "w2", ModifiedTsNs: 3, Fid: &filer_pb.FileId{FileKey: 3}},
|
|
{Offset: 50, Size: 40, FileId: "w3", ModifiedTsNs: 4, Fid: &filer_pb.FileId{FileKey: 4}},
|
|
},
|
|
wantMerge: true,
|
|
},
|
|
{
|
|
name: "overlapping writes with 75pct overlap",
|
|
// Writes of size 100 at step 25 -> each supersedes 75% of the
|
|
// previous write, all survive compact because the leading 25
|
|
// bytes are still visible.
|
|
// total = N*100, file ~ N*25+75. Ratio approaches 4x.
|
|
chunks: func() []*filer_pb.FileChunk {
|
|
var c []*filer_pb.FileChunk
|
|
for i := 0; i < 20; i++ {
|
|
c = append(c, &filer_pb.FileChunk{
|
|
Offset: int64(i * 25),
|
|
Size: 100,
|
|
FileId: string(rune(i)),
|
|
ModifiedTsNs: int64(i + 1),
|
|
Fid: &filer_pb.FileId{FileKey: uint64(i)},
|
|
})
|
|
}
|
|
return c
|
|
}(),
|
|
wantMerge: true,
|
|
},
|
|
{
|
|
name: "many 4K writes at 1K step over large range",
|
|
// 200 writes of 4096 bytes at 1024-byte step.
|
|
// After compact all survive (each partially visible).
|
|
// total = 200*4096 = 819200, file ~ 199*1024+4096 = 207872.
|
|
// ratio ~ 3.9x.
|
|
chunks: func() []*filer_pb.FileChunk {
|
|
var c []*filer_pb.FileChunk
|
|
for i := 0; i < 200; i++ {
|
|
c = append(c, &filer_pb.FileChunk{
|
|
Offset: int64(i * 1024),
|
|
Size: 4096,
|
|
FileId: string(rune(i)),
|
|
ModifiedTsNs: int64(i + 1),
|
|
Fid: &filer_pb.FileId{FileKey: uint64(i)},
|
|
})
|
|
}
|
|
return c
|
|
}(),
|
|
wantMerge: true,
|
|
},
|
|
}
|
|
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
compacted, _ := filer.CompactFileChunks(context.Background(), nil, tt.chunks)
|
|
_, _, merge := shouldMergeChunks(compacted, nil)
|
|
if merge != tt.wantMerge {
|
|
var total uint64
|
|
for _, c := range compacted {
|
|
total += c.Size
|
|
}
|
|
fileSize := filer.TotalSize(compacted)
|
|
t.Fatalf("got merge=%v want %v (compacted=%d total=%d fileSize=%d ratio=%.2f)",
|
|
merge, tt.wantMerge, len(compacted), total, fileSize,
|
|
float64(total)/float64(fileSize))
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
// TestRandomWritesBloatDetection simulates random overwrites and verifies
|
|
// that the condition agrees with the actual stored-vs-visible ratio.
|
|
func TestRandomWritesBloatDetection(t *testing.T) {
|
|
const (
|
|
fileSize = 1024 * 1024 // 1 MB logical file
|
|
chunkSize = 64 * 1024 // 64 KB write size
|
|
)
|
|
|
|
for iter := 0; iter < 50; iter++ {
|
|
chunks := []*filer_pb.FileChunk{
|
|
{Offset: 0, Size: uint64(fileSize), FileId: "base", ModifiedTsNs: 1,
|
|
Fid: &filer_pb.FileId{FileKey: 0}},
|
|
}
|
|
|
|
// Add random overwrites until stored > 3x file size.
|
|
for i := 1; ; i++ {
|
|
off := int64(rand.IntN(fileSize - chunkSize + 1))
|
|
chunks = append(chunks, &filer_pb.FileChunk{
|
|
Offset: off,
|
|
Size: chunkSize,
|
|
FileId: string(rune(i)),
|
|
ModifiedTsNs: int64(1 + i),
|
|
Fid: &filer_pb.FileId{FileKey: uint64(i)},
|
|
})
|
|
var total uint64
|
|
for _, c := range chunks {
|
|
total += c.Size
|
|
}
|
|
if total > uint64(fileSize)*3 {
|
|
break
|
|
}
|
|
}
|
|
|
|
compacted, _ := filer.CompactFileChunks(context.Background(), nil, chunks)
|
|
totalCompacted, reportedFileSize, merge := shouldMergeChunks(compacted, nil)
|
|
|
|
// Sanity: condition must be consistent with its own inputs.
|
|
expectMerge := reportedFileSize > 0 && totalCompacted > 2*reportedFileSize
|
|
if merge != expectMerge {
|
|
t.Fatalf("iter %d: merge=%v but totalCompacted=%d reportedFileSize=%d",
|
|
iter, merge, totalCompacted, reportedFileSize)
|
|
}
|
|
}
|
|
}
|
|
|
|
// 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
|
|
// content as the original.
|
|
func TestVisibleContentPreservedAfterCompact(t *testing.T) {
|
|
const dataSize = 4096
|
|
|
|
for iter := 0; iter < 100; iter++ {
|
|
var chunks []*filer_pb.FileChunk
|
|
numWrites := 5 + rand.IntN(20)
|
|
for i := 0; i < numWrites; i++ {
|
|
start := rand.IntN(dataSize)
|
|
maxSize := 256
|
|
if dataSize-start < maxSize {
|
|
maxSize = dataSize - start
|
|
}
|
|
size := 1 + rand.IntN(maxSize)
|
|
chunks = append(chunks, &filer_pb.FileChunk{
|
|
Offset: int64(start),
|
|
Size: uint64(size),
|
|
FileId: string(rune(i)),
|
|
ModifiedTsNs: int64(i + 1),
|
|
Fid: &filer_pb.FileId{FileKey: uint64(i)},
|
|
})
|
|
}
|
|
|
|
origViews := filer.ViewFromChunks(context.Background(), nil, chunks, 0, math.MaxInt64)
|
|
compacted, _ := filer.CompactFileChunks(context.Background(), nil, chunks)
|
|
compViews := filer.ViewFromChunks(context.Background(), nil, compacted, 0, math.MaxInt64)
|
|
|
|
// Collect all (offset, size) pairs from views.
|
|
type viewKey struct {
|
|
offset int64
|
|
size uint64
|
|
}
|
|
origSet := make(map[viewKey]bool)
|
|
for x := origViews.Front(); x != nil; x = x.Next {
|
|
v := x.Value
|
|
origSet[viewKey{v.ViewOffset, v.ViewSize}] = true
|
|
}
|
|
compSet := make(map[viewKey]bool)
|
|
for x := compViews.Front(); x != nil; x = x.Next {
|
|
v := x.Value
|
|
compSet[viewKey{v.ViewOffset, v.ViewSize}] = true
|
|
}
|
|
|
|
// Every view in the compacted set must exist in the original.
|
|
for k := range compSet {
|
|
if !origSet[k] {
|
|
t.Fatalf("iter %d: compacted has view {offset=%d size=%d} not in original",
|
|
iter, k.offset, k.size)
|
|
}
|
|
}
|
|
// Every view in the original must exist after compaction.
|
|
for k := range origSet {
|
|
if !compSet[k] {
|
|
t.Fatalf("iter %d: original view {offset=%d size=%d} lost after compaction",
|
|
iter, k.offset, k.size)
|
|
}
|
|
}
|
|
}
|
|
}
|