From 7b3462be6a32f131f8905cf0c52baf2b25f9c3a0 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 24 Jul 2026 17:59:20 -0700 Subject: [PATCH] filer: stop an oversized metadata log flush from wedging the change feed (#10430) * filer: write the metadata log in pieces a volume server will accept A single oversized metadata event grows the log buffer past the volume server's fileSizeLimitMB, and the flush of that buffer is then rejected forever: the retry loop has no exit, so the blob at the head of the queue blocks every later flush and the metadata feed stalls until restart. Split the flushed buffer into BufferSize pieces, on record boundaries where possible so each piece still decodes on its own, and retry each piece separately so a partial success is not replayed. Log files are already read as a chunk stream, with a whole-file fallback when a chunk does not decode standalone, so a record may cross a piece boundary. * log_buffer: let go of a window array grown for an oversized entry An entry larger than BufferSize grows the window array to 2*size+4, and window arrays cycle through SealBuffer rather than being freed. One such entry therefore leaves every later window carrying its size, and currentSnapshotView allocates a snapshot as wide as the array on each window, so a few KB of metadata keeps paying for it. Drop the array when SealBuffer hands it back. Growth is on demand, so the next oversized entry just reallocates. * iceberg maintenance: store merged data files as chunks, not inline saveFilerFile had no size threshold, so compaction wrote whole merged parquet files -- hundreds of MB -- as Entry.Content. That puts the parquet bytes verbatim in the filer store and sends them through the metadata change log again as one event. Keep manifests and metadata JSON inline, upload anything larger to volume servers in chunks, assigning through the filer so the path's storage rules apply. * filer: follow the file size limit the volume servers report The starting piece size is a constant, so a cluster whose -fileSizeLimitMB is set below it would reject every piece and wedge just the same. The rejection names the limit, so take it from there and re-cut the rest of the flush to fit. Piece the buffer one at a time rather than up front, since the size can change partway through a flush. --- weed/filer/filer_notify.go | 69 ++++++- weed/filer/filer_notify_test.go | 162 +++++++++++++++++ weed/util/log_buffer/log_buffer.go | 8 + .../log_buffer/log_buffer_oversized_test.go | 66 +++++++ weed/worker/tasks/iceberg/exec_test.go | 18 ++ weed/worker/tasks/iceberg/filer_io.go | 93 ++++++++-- .../tasks/iceberg/filer_io_save_test.go | 170 ++++++++++++++++++ 7 files changed, 572 insertions(+), 14 deletions(-) create mode 100644 weed/util/log_buffer/log_buffer_oversized_test.go create mode 100644 weed/worker/tasks/iceberg/filer_io_save_test.go diff --git a/weed/filer/filer_notify.go b/weed/filer/filer_notify.go index 19f020185..a9ba55438 100644 --- a/weed/filer/filer_notify.go +++ b/weed/filer/filer_notify.go @@ -7,6 +7,7 @@ import ( "io" nethttp "net/http" "regexp" + "strconv" "strings" "time" @@ -155,6 +156,32 @@ func (f *Filer) triggerLocalEmptyFolderCleanup(oldEntry, newEntry *Entry) { } } +// metadataLogUploadLimit is the piece size a metadata log flush starts with. A +// volume server refuses anything over its -fileSizeLimitMB (256 MB by default), +// and a single oversized event — a CreateEntry carrying a large inline Content, +// say — grows the log buffer well past that, leaving a blob that can never be +// written and blocks every later flush behind it. BufferSize is what an +// ordinary flush already produces, so it is a size the volume server accepts +// under any configuration that works at all; a cluster running below it says so +// in the rejection and volumeFileSizeLimit picks the real limit up from there. +const metadataLogUploadLimit = log_buffer.BufferSize + +var fileSizeLimitPattern = regexp.MustCompile(`file over the limited (\d+) bytes`) + +// volumeFileSizeLimit reads the byte limit back out of a volume server's size +// rejection, and returns 0 for any other error. +func volumeFileSizeLimit(err error) int { + match := fileSizeLimitPattern.FindStringSubmatch(err.Error()) + if match == nil { + return 0 + } + limit, convErr := strconv.Atoi(match[1]) + if convErr != nil { + return 0 + } + return limit +} + func (f *Filer) logFlushFunc(logBuffer *log_buffer.LogBuffer, startTime, stopTime time.Time, buf []byte, minOffset, maxOffset int64) { if len(buf) == 0 { @@ -168,14 +195,50 @@ func (f *Filer) logFlushFunc(logBuffer *log_buffer.LogBuffer, startTime, stopTim // startTime.Second(), startTime.Nanosecond(), ) - for { - if err := f.appendToFile(targetFile, buf); err != nil { + // One piece at a time, each retried on its own so a partial success is not + // replayed, and the piece size follows the limit the volume servers report. + limit := metadataLogUploadLimit + for len(buf) > 0 { + piece := nextLogPiece(buf, limit) + if err := f.appendToFile(targetFile, piece); err != nil { glog.V(0).Infof("metadata log write failed %s: %v", targetFile, err) + if reported := volumeFileSizeLimit(err); reported > 0 && reported < limit { + glog.V(0).Infof("metadata log upload limit lowered to %d bytes", reported) + limit = reported + continue + } time.Sleep(737 * time.Millisecond) - } else { + continue + } + buf = buf[len(piece):] + } +} + +// nextLogPiece returns the leading piece of a flushed log buffer, at most +// maxSize bytes and ending on a record boundary where it can so the piece still +// decodes on its own. A record longer than maxSize is cut by size instead; the +// readers fall back to streaming the whole file when a chunk does not decode +// standalone, so a record may cross a chunk boundary. +func nextLogPiece(buf []byte, maxSize int) []byte { + if len(buf) <= maxSize { + return buf + } + + pos := 0 + for pos+4 <= len(buf) { + size := int(util.BytesToUint32(buf[pos : pos+4])) + end := pos + 4 + size + if size <= 0 || end > len(buf) || end > maxSize { break } + pos = end } + if pos == 0 { + // Either the leading record alone is over the limit, or buf starts + // mid-record because the piece before it was cut by size. + return buf[:maxSize] + } + return buf[:pos] } var ( diff --git a/weed/filer/filer_notify_test.go b/weed/filer/filer_notify_test.go index af99d7015..cfe5651d7 100644 --- a/weed/filer/filer_notify_test.go +++ b/weed/filer/filer_notify_test.go @@ -1,6 +1,9 @@ package filer import ( + "bytes" + "errors" + "fmt" "testing" "time" @@ -51,3 +54,162 @@ func TestProtoMarshal(t *testing.T) { println(string(text)) } + +// buildLogBuffer lays out records exactly as LogBuffer.AddDataToBuffer does: +// a 4-byte size prefix in front of each marshaled LogEntry. +func buildLogBuffer(t *testing.T, payloadSizes []int) (buf []byte, count int) { + t.Helper() + sizeBuf := make([]byte, 4) + for i, payloadSize := range payloadSizes { + data, err := proto.Marshal(&filer_pb.LogEntry{ + TsNs: int64(i + 1), + Data: make([]byte, payloadSize), + }) + if err != nil { + t.Fatalf("marshal log entry: %v", err) + } + util.Uint32toBytes(sizeBuf, uint32(len(data))) + buf = append(buf, sizeBuf...) + buf = append(buf, data...) + } + return buf, len(payloadSizes) +} + +// splitLogBuffer drains a buffer the way logFlushFunc does, one piece at a +// time, so the tests exercise the same walk. +func splitLogBuffer(buf []byte, maxSize int) [][]byte { + var pieces [][]byte + for len(buf) > 0 { + piece := nextLogPiece(buf, maxSize) + pieces = append(pieces, piece) + buf = buf[len(piece):] + } + return pieces +} + +func TestNextLogPiece(t *testing.T) { + const maxSize = 1024 + + testCases := []struct { + name string + payloadSizes []int + }{ + {"fits in one piece", []int{10, 20, 30}}, + {"many small records", []int{300, 300, 300, 300, 300, 300, 300}}, + {"one record far over the limit", []int{5000}}, + {"oversized record between small ones", []int{100, 5000, 100}}, + {"record exactly at the limit", []int{maxSize - 4 - 6}}, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + buf, count := buildLogBuffer(t, tc.payloadSizes) + + var rejoined []byte + for i, piece := range splitLogBuffer(buf, maxSize) { + if len(piece) > maxSize { + t.Errorf("piece %d is %d bytes, over the %d limit", i, len(piece), maxSize) + } + if len(piece) == 0 { + t.Errorf("piece %d is empty", i) + } + rejoined = append(rejoined, piece...) + } + if !bytes.Equal(rejoined, buf) { + t.Errorf("rejoined pieces differ from the original buffer") + } + + // The pieces still stream back as the same records in order. + decoded, _, err := decodeLogRecords(rejoined) + if err != nil { + t.Fatalf("decode rejoined buffer: %v", err) + } + if len(decoded) != count { + t.Errorf("decoded %d records, want %d", len(decoded), count) + } + }) + } +} + +// A buffer of ordinary records splits on record boundaries, so every piece +// decodes on its own and the readers keep the per-chunk cache path. +func TestNextLogPieceKeepsRecordBoundaries(t *testing.T) { + buf, count := buildLogBuffer(t, []int{300, 300, 300, 300, 300, 300, 300}) + + var decodedCount int + for i, piece := range splitLogBuffer(buf, 1024) { + entries, cacheable, err := decodeLogRecords(piece) + if err != nil { + t.Fatalf("piece %d does not decode standalone: %v", i, err) + } + if !cacheable { + t.Errorf("piece %d is not cacheable", i) + } + decodedCount += len(entries) + } + if decodedCount != count { + t.Errorf("decoded %d records across pieces, want %d", decodedCount, count) + } +} + +// A truncated tail must not be dropped or duplicated, only cut by size. +func TestNextLogPieceTruncatedTail(t *testing.T) { + buf, _ := buildLogBuffer(t, []int{300, 300}) + buf = append(buf, 0xff, 0xff, 0xff) + + var rejoined []byte + for i, piece := range splitLogBuffer(buf, 320) { + if len(piece) > 320 { + t.Errorf("piece %d is %d bytes, over the 320 limit", i, len(piece)) + } + rejoined = append(rejoined, piece...) + } + if !bytes.Equal(rejoined, buf) { + t.Errorf("rejoined pieces differ from the original buffer") + } +} + +// A cluster whose fileSizeLimitMB is under metadataLogUploadLimit says so in +// the rejection, and the flush has to take that limit up rather than retry an +// unwritable piece forever. +func TestVolumeFileSizeLimit(t *testing.T) { + testCases := []struct { + name string + err error + want int + }{ + { + "volume server size rejection", + fmt.Errorf("upload data http://127.0.0.1:8180/1,16ef8dca8a: unmarshalled error http://127.0.0.1:8180/1,16ef8dca8a: file over the limited 16777216 bytes"), + 16777216, + }, + {"default limit", errors.New("file over the limited 268435456 bytes"), 268435456}, + {"unrelated failure", errors.New("connection refused"), 0}, + {"no byte count", errors.New("file over the limited bytes"), 0}, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + if got := volumeFileSizeLimit(tc.err); got != tc.want { + t.Errorf("volumeFileSizeLimit = %d, want %d", got, tc.want) + } + }) + } +} + +// Once the reported limit is adopted, every piece fits it. +func TestNextLogPieceHonorsLoweredLimit(t *testing.T) { + buf, _ := buildLogBuffer(t, []int{20 << 20}) + + const lowered = 1 << 20 + var rejoined []byte + for i, piece := range splitLogBuffer(buf, lowered) { + if len(piece) > lowered { + t.Errorf("piece %d is %d bytes, over the lowered %d limit", i, len(piece), lowered) + } + rejoined = append(rejoined, piece...) + } + if !bytes.Equal(rejoined, buf) { + t.Errorf("rejoined pieces differ from the original buffer") + } +} diff --git a/weed/util/log_buffer/log_buffer.go b/weed/util/log_buffer/log_buffer.go index d6fd2b50e..6d2970a83 100644 --- a/weed/util/log_buffer/log_buffer.go +++ b/weed/util/log_buffer/log_buffer.go @@ -616,6 +616,14 @@ func (logBuffer *LogBuffer) copyToFlushInternal(withCallback bool) *dataToFlush // CRITICAL: logBuffer.offset is the "next offset to assign", so last offset in buffer is offset-1 lastOffsetInBuffer := logBuffer.offset - 1 logBuffer.buf = logBuffer.prevBuffers.SealBuffer(logBuffer.startTime, logBuffer.stopTime, logBuffer.buf, logBuffer.pos, logBuffer.bufferStartOffset, lastOffsetInBuffer) + // SealBuffer hands back the oldest window array to reuse. An entry larger + // than BufferSize grew one of these arrays to fit it, and buffers cycle + // forever, so without this a single oversized entry leaves every later + // window carrying — and snapshotting — its size. Growth is on demand, so + // the next oversized entry just reallocates. + if len(logBuffer.buf) > BufferSize { + logBuffer.buf = make([]byte, BufferSize) + } // Hand a fully extended prefix snapshot to the sealed slot so sealed // readers reuse it instead of re-copying the window; reset for the next // window either way (holders keep their immutable prefix slices). diff --git a/weed/util/log_buffer/log_buffer_oversized_test.go b/weed/util/log_buffer/log_buffer_oversized_test.go new file mode 100644 index 000000000..c5deb3b73 --- /dev/null +++ b/weed/util/log_buffer/log_buffer_oversized_test.go @@ -0,0 +1,66 @@ +package log_buffer + +import ( + "testing" + "time" +) + +// An entry larger than BufferSize grows the window array to hold it. Window +// arrays cycle through SealBuffer rather than being freed, so the grown one has +// to be let go once it comes back around -- otherwise every later window is +// allocated and snapshotted at the oversized width. +func TestOversizedEntryDoesNotInflateBufferForever(t *testing.T) { + lb := NewLogBuffer("oversized", time.Hour, func(_ *LogBuffer, _, _ time.Time, _ []byte, _, _ int64) {}, nil, func() {}) + defer lb.ShutdownLogBuffer() + + if err := lb.AddDataToBuffer(nil, make([]byte, BufferSize+1), time.Now().UnixNano()); err != nil { + t.Fatalf("add oversized entry: %v", err) + } + if len(lb.buf) <= BufferSize { + t.Fatalf("oversized entry did not grow the buffer: %d bytes", len(lb.buf)) + } + + // One seal per window: the grown array lands in the newest sealed slot and + // needs a full trip through the rotation to come back as the current buffer. + for i := 0; i < PreviousBufferCount+2; i++ { + lb.ForceFlush() + if err := lb.AddDataToBuffer(nil, []byte("small"), time.Now().UnixNano()); err != nil { + t.Fatalf("add small entry: %v", err) + } + } + + if len(lb.buf) > BufferSize { + t.Errorf("current buffer still %d bytes, want at most %d", len(lb.buf), BufferSize) + } + for i, sealed := range lb.prevBuffers.buffers { + if len(sealed.buf) > BufferSize { + t.Errorf("sealed buffer %d still %d bytes, want at most %d", i, len(sealed.buf), BufferSize) + } + } +} + +// The oversized entry itself still has to survive the round trip. +func TestOversizedEntryStillFlushes(t *testing.T) { + var flushed [][]byte + lb := NewLogBuffer("oversized-flush", time.Hour, func(_ *LogBuffer, _, _ time.Time, buf []byte, _, _ int64) { + flushed = append(flushed, append([]byte(nil), buf...)) + }, nil, func() {}) + defer lb.ShutdownLogBuffer() + + payload := make([]byte, BufferSize+1) + for i := range payload { + payload[i] = byte(i) + } + if err := lb.AddDataToBuffer(nil, payload, time.Now().UnixNano()); err != nil { + t.Fatalf("add oversized entry: %v", err) + } + lb.ForceFlush() + + var total int + for _, buf := range flushed { + total += len(buf) + } + if total < len(payload) { + t.Errorf("flushed %d bytes, want at least the %d byte payload", total, len(payload)) + } +} diff --git a/weed/worker/tasks/iceberg/exec_test.go b/weed/worker/tasks/iceberg/exec_test.go index 01881e474..6495bd716 100644 --- a/weed/worker/tasks/iceberg/exec_test.go +++ b/weed/worker/tasks/iceberg/exec_test.go @@ -41,6 +41,10 @@ type fakeFilerServer struct { entries map[string]map[string]*filer_pb.Entry // dir → name → entry beforeUpdate func(*fakeFilerServer, *filer_pb.UpdateEntryRequest) error + // Set by enableAssign to serve AssignVolume against a fake volume server. + assignVolumeServer string + assignCount int + // Counters for assertions createCalls int updateCalls int @@ -190,6 +194,20 @@ func (f *fakeFilerServer) DeleteEntry(_ context.Context, req *filer_pb.DeleteEnt return &filer_pb.DeleteEntryResponse{}, nil } +func (f *fakeFilerServer) AssignVolume(_ context.Context, _ *filer_pb.AssignVolumeRequest) (*filer_pb.AssignVolumeResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + if f.assignVolumeServer == "" { + return nil, status.Error(codes.Unavailable, "no volume server") + } + f.assignCount++ + return &filer_pb.AssignVolumeResponse{ + FileId: fmt.Sprintf("1,%08x", f.assignCount), + Location: &filer_pb.Location{Url: f.assignVolumeServer, PublicUrl: f.assignVolumeServer}, + Count: 1, + }, nil +} + func (f *fakeFilerServer) Ping(_ context.Context, _ *filer_pb.PingRequest) (*filer_pb.PingResponse, error) { now := time.Now().UnixNano() return &filer_pb.PingResponse{ diff --git a/weed/worker/tasks/iceberg/filer_io.go b/weed/worker/tasks/iceberg/filer_io.go index 4f03f3337..7564659eb 100644 --- a/weed/worker/tasks/iceberg/filer_io.go +++ b/weed/worker/tasks/iceberg/filer_io.go @@ -5,6 +5,7 @@ import ( "crypto/rand" "encoding/hex" "encoding/json" + "errors" "fmt" "io" "path" @@ -15,8 +16,11 @@ import ( "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/operation" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" + "github.com/seaweedfs/seaweedfs/weed/security" + "github.com/seaweedfs/seaweedfs/weed/util" util_http "github.com/seaweedfs/seaweedfs/weed/util/http" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" @@ -252,20 +256,42 @@ func absoluteIcebergPath(bucketName string, elem ...string) string { return "s3://" + path.Join(append([]string{bucketName}, elem...)...) } -// saveFilerFile saves a file to the filer. +const ( + // inlineContentLimit is the most saveFilerFile keeps in Entry.Content. + // Manifests and metadata JSON are small and stay inline; a compacted data + // file runs to hundreds of MB, and inlining that stores the parquet bytes + // verbatim in the filer store and ships them again through the metadata + // change log as one oversized event. + inlineContentLimit = 256 * 1024 + // filerFileChunkSize is the upload chunk size for anything over the limit. + filerFileChunkSize = 8 * 1024 * 1024 +) + +// saveFilerFile saves a file to the filer, inline when it is small and as +// volume chunks when it is not. func saveFilerFile(ctx context.Context, client filer_pb.SeaweedFilerClient, dir, fileName string, content []byte) error { + entry := &filer_pb.Entry{ + Name: fileName, + Attributes: &filer_pb.FuseAttributes{ + Mtime: time.Now().Unix(), + Crtime: time.Now().Unix(), + FileMode: uint32(0644), + FileSize: uint64(len(content)), + }, + } + if len(content) <= inlineContentLimit { + entry.Content = content + } else { + chunks, err := uploadFilerChunks(ctx, client, path.Join(dir, fileName), content) + if err != nil { + return fmt.Errorf("upload %s/%s: %w", dir, fileName, err) + } + entry.Chunks = chunks + } + resp, err := client.CreateEntry(ctx, &filer_pb.CreateEntryRequest{ Directory: dir, - Entry: &filer_pb.Entry{ - Name: fileName, - Attributes: &filer_pb.FuseAttributes{ - Mtime: time.Now().Unix(), - Crtime: time.Now().Unix(), - FileMode: uint32(0644), - FileSize: uint64(len(content)), - }, - Content: content, - }, + Entry: entry, }) if err != nil { return fmt.Errorf("create entry %s/%s: %w", dir, fileName, err) @@ -276,6 +302,51 @@ func saveFilerFile(ctx context.Context, client filer_pb.SeaweedFilerClient, dir, return nil } +// uploadFilerChunks writes content to volume servers, assigning through the +// filer so the entry's storage rules apply. +func uploadFilerChunks(ctx context.Context, client filer_pb.SeaweedFilerClient, fullPath string, content []byte) ([]*filer_pb.FileChunk, error) { + assignFn := func(ctx context.Context, count int, expectedDataSize uint64) (*operation.VolumeAssignRequest, *operation.AssignResult, error) { + resp, err := client.AssignVolume(ctx, &filer_pb.AssignVolumeRequest{ + Count: int32(count), + Path: fullPath, + ExpectedDataSize: expectedDataSize, + }) + if err != nil { + return nil, nil, err + } + if resp.Error != "" { + return nil, nil, errors.New(resp.Error) + } + if resp.Location == nil || resp.FileId == "" { + return nil, nil, fmt.Errorf("assign volume returned no location") + } + return nil, &operation.AssignResult{ + Fid: resp.FileId, + Url: resp.Location.Url, + PublicUrl: resp.Location.PublicUrl, + Count: uint64(count), + Auth: security.EncodedJwt(resp.Auth), + }, nil + } + + initGlobalHTTPClientOnce.Do(util_http.InitGlobalHttpClient) + result, err := operation.UploadReaderInChunks(ctx, util.NewBytesReader(content), &operation.ChunkedUploadOption{ + ChunkSize: filerFileChunkSize, + AssignFunc: assignFn, + }) + if err != nil { + // Chunks uploaded before the failure are orphaned: nothing references + // them, so name them for fsck rather than losing them silently. + if result != nil { + for _, chunk := range result.FileChunks { + glog.Warningf("iceberg: orphan chunk %s from failed upload of %s", chunk.GetFileIdString(), fullPath) + } + } + return nil, err + } + return result.FileChunks, nil +} + // deleteFilerFile deletes a file from the filer. func deleteFilerFile(ctx context.Context, client filer_pb.SeaweedFilerClient, dir, fileName string) error { return filer_pb.DoRemove(ctx, client, dir, fileName, true, false, true, false, nil) diff --git a/weed/worker/tasks/iceberg/filer_io_save_test.go b/weed/worker/tasks/iceberg/filer_io_save_test.go new file mode 100644 index 000000000..2040f11e5 --- /dev/null +++ b/weed/worker/tasks/iceberg/filer_io_save_test.go @@ -0,0 +1,170 @@ +package iceberg + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "sync" + "testing" + + util_http "github.com/seaweedfs/seaweedfs/weed/util/http" +) + +// fakeVolumeServer accepts chunk uploads and records what it received, so a +// test can check the bytes actually left the filer entry. +type fakeVolumeServer struct { + mu sync.Mutex + server *httptest.Server + parts map[string][]byte // fid → data +} + +func startFakeVolumeServer(t *testing.T) *fakeVolumeServer { + t.Helper() + initGlobalHTTPClientOnce.Do(util_http.InitGlobalHttpClient) + + v := &fakeVolumeServer{parts: make(map[string][]byte)} + v.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + fid := strings.TrimPrefix(r.URL.Path, "/") + // The uploader sends the payload as a single form-data part named + // "file" with an empty filename, which FormFile will not match. + reader, err := r.MultipartReader() + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + part, err := reader.NextPart() + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + data, err := io.ReadAll(part) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + v.mu.Lock() + v.parts[fid] = data + v.mu.Unlock() + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{"name": fid, "size": len(data)}) + })) + t.Cleanup(v.server.Close) + return v +} + +func (v *fakeVolumeServer) hostPort(t *testing.T) string { + t.Helper() + u, err := url.Parse(v.server.URL) + if err != nil { + t.Fatalf("parse fake volume url: %v", err) + } + return u.Host +} + +func (v *fakeVolumeServer) uploaded() int { + v.mu.Lock() + defer v.mu.Unlock() + return len(v.parts) +} + +func (v *fakeVolumeServer) dataFor(fid string) []byte { + v.mu.Lock() + defer v.mu.Unlock() + return v.parts[fid] +} + +// AssignVolume on the fake filer hands out sequential fids pointing at the +// fake volume server. +func (f *fakeFilerServer) enableAssign(volumeServer string) { + f.mu.Lock() + defer f.mu.Unlock() + f.assignVolumeServer = volumeServer +} + +func TestSaveFilerFileKeepsSmallContentInline(t *testing.T) { + fakeServer, client := startFakeFiler(t) + + content := []byte(strings.Repeat("m", inlineContentLimit)) + if err := saveFilerFile(context.Background(), client, "/buckets/lake/ns/tbl/metadata", "snap.avro", content); err != nil { + t.Fatalf("saveFilerFile: %v", err) + } + + entry := fakeServer.getEntry("/buckets/lake/ns/tbl/metadata", "snap.avro") + if entry == nil { + t.Fatal("entry not created") + } + if len(entry.Content) != len(content) { + t.Errorf("inline content is %d bytes, want %d", len(entry.Content), len(content)) + } + if len(entry.Chunks) != 0 { + t.Errorf("small file got %d chunks, want inline only", len(entry.Chunks)) + } +} + +// A compacted data file must land in volumes, not in Entry.Content: an inline +// entry that big is stored verbatim by the filer store and rides the metadata +// change log as one event too large for a volume server to accept. +func TestSaveFilerFileChunksLargeContent(t *testing.T) { + fakeServer, client := startFakeFiler(t) + volumeServer := startFakeVolumeServer(t) + fakeServer.enableAssign(volumeServer.hostPort(t)) + + content := make([]byte, 3*filerFileChunkSize/2) + for i := range content { + content[i] = byte(i) + } + + if err := saveFilerFile(context.Background(), client, "/buckets/lake/ns/tbl/data", "compact-1.parquet", content); err != nil { + t.Fatalf("saveFilerFile: %v", err) + } + + entry := fakeServer.getEntry("/buckets/lake/ns/tbl/data", "compact-1.parquet") + if entry == nil { + t.Fatal("entry not created") + } + if len(entry.Content) != 0 { + t.Errorf("large file kept %d bytes inline, want none", len(entry.Content)) + } + if len(entry.Chunks) != 2 { + t.Fatalf("got %d chunks, want 2", len(entry.Chunks)) + } + if entry.Attributes.FileSize != uint64(len(content)) { + t.Errorf("file size %d, want %d", entry.Attributes.FileSize, len(content)) + } + if volumeServer.uploaded() != 2 { + t.Errorf("volume server received %d uploads, want 2", volumeServer.uploaded()) + } + + // Reassembling the chunks in offset order gives back the original bytes. + rejoined := make([]byte, len(content)) + for _, chunk := range entry.Chunks { + data := volumeServer.dataFor(chunk.FileId) + if uint64(len(data)) != chunk.Size { + t.Fatalf("chunk %s holds %d bytes, entry says %d", chunk.FileId, len(data), chunk.Size) + } + copy(rejoined[chunk.Offset:], data) + } + if string(rejoined) != string(content) { + t.Error("reassembled chunks differ from the original content") + } +} + +// A failed upload must not silently fall back to an inline entry. +func TestSaveFilerFileFailsWhenUploadFails(t *testing.T) { + fakeServer, client := startFakeFiler(t) + + content := make([]byte, filerFileChunkSize+1) + err := saveFilerFile(context.Background(), client, "/buckets/lake/ns/tbl/data", "compact-2.parquet", content) + if err == nil { + t.Fatal("expected an error when no volume can be assigned") + } + if entry := fakeServer.getEntry("/buckets/lake/ns/tbl/data", "compact-2.parquet"); entry != nil { + t.Errorf("entry created despite the failed upload: %d inline bytes", len(entry.Content)) + } +}