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)) + } +}