diff --git a/weed/filer/filechunk_manifest_test.go b/weed/filer/filechunk_manifest_test.go index 0cec6c8d1..e23bf5366 100644 --- a/weed/filer/filechunk_manifest_test.go +++ b/weed/filer/filechunk_manifest_test.go @@ -338,7 +338,10 @@ func TestCompactResolvedOverlappingManifests(t *testing.T) { } // CompactFileChunks on resolved sub-chunks (nil lookupFn is fine for non-manifest chunks) - compacted, garbage := CompactFileChunks(context.Background(), nil, dataChunks) + compacted, garbage, err := CompactFileChunks(context.Background(), nil, dataChunks) + if err != nil { + t.Fatal(err) + } if len(compacted) != 3 { t.Fatalf("expected 3 compacted chunks, got %d", len(compacted)) @@ -487,7 +490,10 @@ func TestCompactMultipleOverlappingManifestGenerations(t *testing.T) { } // Compact: only generation 5 (the newest 3 chunks) should survive - compacted, garbage := CompactFileChunks(context.Background(), nil, dataChunks) + compacted, garbage, err := CompactFileChunks(context.Background(), nil, dataChunks) + if err != nil { + t.Fatal(err) + } if len(compacted) != 3 { t.Errorf("expected 3 compacted (gen 5), got %d", len(compacted)) } diff --git a/weed/filer/filechunks.go b/weed/filer/filechunks.go index 954547296..3a61ed3b5 100644 --- a/weed/filer/filechunks.go +++ b/weed/filer/filechunks.go @@ -63,9 +63,15 @@ func ETagChunks(chunks []*filer_pb.FileChunk) (etag string) { return finalETag } -func CompactFileChunks(ctx context.Context, lookupFileIdFn wdclient.LookupFileIdFunctionType, chunks []*filer_pb.FileChunk) (compacted, garbage []*filer_pb.FileChunk) { +// CompactFileChunks splits chunks into those still visible and those fully +// covered by newer writes. It fails if the chunk manifests cannot be resolved, +// including when ctx is cancelled. +func CompactFileChunks(ctx context.Context, lookupFileIdFn wdclient.LookupFileIdFunctionType, chunks []*filer_pb.FileChunk) (compacted, garbage []*filer_pb.FileChunk, err error) { - visibles, _ := NonOverlappingVisibleIntervals(ctx, lookupFileIdFn, chunks, 0, math.MaxInt64) + visibles, err := NonOverlappingVisibleIntervals(ctx, lookupFileIdFn, chunks, 0, math.MaxInt64) + if err != nil { + return nil, nil, err + } compacted, garbage = SeparateGarbageChunks(visibles, chunks) @@ -194,11 +200,17 @@ func (cv *ChunkView) CanRangeFetch() bool { return cv.CipherKey == nil && !cv.IsGzipped } -func ViewFromChunks(ctx context.Context, lookupFileIdFn wdclient.LookupFileIdFunctionType, chunks []*filer_pb.FileChunk, offset int64, size int64) (chunkViews *IntervalList[*ChunkView]) { +// ViewFromChunks returns the views of chunks that cover [offset, offset+size). +// It fails if the chunk manifests cannot be resolved, including when ctx is +// cancelled. +func ViewFromChunks(ctx context.Context, lookupFileIdFn wdclient.LookupFileIdFunctionType, chunks []*filer_pb.FileChunk, offset int64, size int64) (chunkViews *IntervalList[*ChunkView], err error) { - visibles, _ := NonOverlappingVisibleIntervals(ctx, lookupFileIdFn, chunks, offset, offset+size) + visibles, err := NonOverlappingVisibleIntervals(ctx, lookupFileIdFn, chunks, offset, offset+size) + if err != nil { + return nil, err + } - return ViewFromVisibleIntervals(visibles, offset, size) + return ViewFromVisibleIntervals(visibles, offset, size), nil } diff --git a/weed/filer/filechunks2_test.go b/weed/filer/filechunks2_test.go index 9c8a916e6..e8b1f39d2 100644 --- a/weed/filer/filechunks2_test.go +++ b/weed/filer/filechunks2_test.go @@ -66,7 +66,10 @@ func TestCompactFileChunksRealCase(t *testing.T) { printChunks("before", chunks) - compacted, garbage := CompactFileChunks(context.Background(), nil, chunks) + compacted, garbage, err := CompactFileChunks(context.Background(), nil, chunks) + if err != nil { + t.Fatal(err) + } printChunks("compacted", compacted) printChunks("garbage", garbage) diff --git a/weed/filer/filechunks_cancel_test.go b/weed/filer/filechunks_cancel_test.go new file mode 100644 index 000000000..689f9478d --- /dev/null +++ b/weed/filer/filechunks_cancel_test.go @@ -0,0 +1,58 @@ +package filer + +import ( + "context" + "errors" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" +) + +// cancelledChunks returns a cancelled context and plain data chunks. Resolving +// chunks under a cancelled context fails even without manifests, so these +// exercise the error path with no lookup function. +func cancelledChunks() (context.Context, []*filer_pb.FileChunk) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + return ctx, []*filer_pb.FileChunk{ + {FileId: "1,01", Offset: 0, Size: 100, ModifiedTsNs: 1}, + {FileId: "1,02", Offset: 0, Size: 100, ModifiedTsNs: 2}, + } +} + +func TestCompactFileChunksCancelledContext(t *testing.T) { + ctx, chunks := cancelledChunks() + + compacted, garbage, err := CompactFileChunks(ctx, nil, chunks) + if !errors.Is(err, context.Canceled) { + t.Fatalf("err = %v, want context.Canceled", err) + } + if compacted != nil || garbage != nil { + t.Fatalf("compacted = %v, garbage = %v, want both nil on error", compacted, garbage) + } +} + +func TestViewFromChunksCancelledContext(t *testing.T) { + ctx, chunks := cancelledChunks() + + views, err := ViewFromChunks(ctx, nil, chunks, 0, 100) + if !errors.Is(err, context.Canceled) { + t.Fatalf("err = %v, want context.Canceled", err) + } + if views != nil { + t.Fatalf("views = %v, want nil on error", views) + } +} + +func TestChunkStreamReaderCancelledContext(t *testing.T) { + ctx, chunks := cancelledChunks() + + reader := NewChunkStreamReaderFromLookup(ctx, nil, chunks) + n, err := reader.Read(make([]byte, 10)) + if n != 0 || !errors.Is(err, context.Canceled) { + t.Fatalf("Read = %d, %v, want 0, context.Canceled", n, err) + } + if !errors.Is(reader.SourceError(), context.Canceled) { + t.Fatalf("SourceError = %v, want context.Canceled", reader.SourceError()) + } +} diff --git a/weed/filer/filechunks_test.go b/weed/filer/filechunks_test.go index 4ae7d6133..716e077a3 100644 --- a/weed/filer/filechunks_test.go +++ b/weed/filer/filechunks_test.go @@ -22,7 +22,10 @@ func TestCompactFileChunks(t *testing.T) { {Offset: 110, Size: 200, FileId: "jkl", ModifiedTsNs: 300}, } - compacted, garbage := CompactFileChunks(context.Background(), nil, chunks) + compacted, garbage, err := CompactFileChunks(context.Background(), nil, chunks) + if err != nil { + t.Fatal(err) + } if len(compacted) != 3 { t.Fatalf("unexpected compacted: %d", len(compacted)) @@ -55,7 +58,10 @@ func TestCompactFileChunks2(t *testing.T) { }) } - compacted, garbage := CompactFileChunks(context.Background(), nil, chunks) + compacted, garbage, err := CompactFileChunks(context.Background(), nil, chunks) + if err != nil { + t.Fatal(err) + } if len(compacted) != 4 { t.Fatalf("unexpected compacted: %d", len(compacted)) @@ -427,7 +433,10 @@ func TestChunksReading(t *testing.T) { // continue } log.Printf("++++++++++ read test case %d ++++++++++++++++++++", i) - chunks := ViewFromChunks(context.Background(), nil, testcase.Chunks, testcase.Offset, testcase.Size) + chunks, err := ViewFromChunks(context.Background(), nil, testcase.Chunks, testcase.Offset, testcase.Size) + if err != nil { + t.Fatal(err) + } x := -1 for c := chunks.Front(); c != nil; c = c.Next { x++ @@ -474,7 +483,9 @@ func BenchmarkCompactFileChunks(b *testing.B) { } for n := 0; n < b.N; n++ { - CompactFileChunks(context.Background(), nil, chunks) + if _, _, err := CompactFileChunks(context.Background(), nil, chunks); err != nil { + b.Fatal(err) + } } } @@ -563,7 +574,10 @@ func TestCompactFileChunks3(t *testing.T) { {Offset: 300, Size: 100, FileId: "def", ModifiedTsNs: 200}, } - compacted, _ := CompactFileChunks(context.Background(), nil, chunks) + compacted, _, err := CompactFileChunks(context.Background(), nil, chunks) + if err != nil { + t.Fatal(err) + } if len(compacted) != 4 { t.Fatalf("unexpected compacted: %d", len(compacted)) diff --git a/weed/filer/stream.go b/weed/filer/stream.go index 47fccbb87..a34a6c548 100644 --- a/weed/filer/stream.go +++ b/weed/filer/stream.go @@ -386,9 +386,18 @@ var _ = io.ReadSeeker(&ChunkStreamReader{}) var _ = io.ReaderAt(&ChunkStreamReader{}) var _ = io.Closer(&ChunkStreamReader{}) +// doNewChunkStreamReader builds a reader over chunks. If their manifests cannot +// be resolved, the reader fails every read with that error and reports it as its +// SourceError. func doNewChunkStreamReader(ctx context.Context, lookupFileIdFn wdclient.LookupFileIdFunctionType, chunks []*filer_pb.FileChunk) *ChunkStreamReader { - chunkViews := ViewFromChunks(ctx, lookupFileIdFn, chunks, 0, math.MaxInt64) + chunkViews, err := ViewFromChunks(ctx, lookupFileIdFn, chunks, 0, math.MaxInt64) + if err != nil { + return &ChunkStreamReader{ + lookupFileId: lookupFileIdFn, + sourceErr: err, + } + } var totalSize int64 for x := chunkViews.Front(); x != nil; x = x.Next { @@ -487,6 +496,10 @@ func insideChunk(offset int64, chunk *ChunkView) bool { } func (c *ChunkStreamReader) prepareBufferFor(offset int64) (err error) { + // a reader without chunk views has a source error only if ViewFromChunks failed + if c.head == nil && c.sourceErr != nil { + return c.sourceErr + } // stay in the same chunk if c.bufferOffset <= offset && offset < c.bufferOffset+int64(len(c.buffer)) { return nil diff --git a/weed/mount/weedfs_file_sync.go b/weed/mount/weedfs_file_sync.go index 834e3827c..8c7d85242 100644 --- a/weed/mount/weedfs_file_sync.go +++ b/weed/mount/weedfs_file_sync.go @@ -238,7 +238,10 @@ func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, n manifestChunks, nonManifestChunks := filer.SeparateManifestChunks(entry.GetChunks()) - chunks, _ := filer.CompactFileChunks(context.Background(), wfs.LookupFn(), nonManifestChunks) + chunks, _, err := filer.CompactFileChunks(context.Background(), wfs.LookupFn(), nonManifestChunks) + if err != nil { + return fmt.Errorf("compact chunks of %s: %w", fileFullPath, err) + } if mergedChunks, mergeErr := wfs.maybeMergeChunks(fileFullPath, chunks, manifestChunks); mergeErr != nil { glog.V(0).Infof("maybeMergeChunks %s: %v", fileFullPath, mergeErr) diff --git a/weed/mount/weedfs_file_sync_test.go b/weed/mount/weedfs_file_sync_test.go index ac8c51872..ed1ce122a 100644 --- a/weed/mount/weedfs_file_sync_test.go +++ b/weed/mount/weedfs_file_sync_test.go @@ -245,7 +245,10 @@ func TestCompactThenMergeCondition(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - compacted, _ := filer.CompactFileChunks(context.Background(), nil, tt.chunks) + compacted, _, err := filer.CompactFileChunks(context.Background(), nil, tt.chunks) + if err != nil { + t.Fatal(err) + } _, _, merge := shouldMergeChunks(compacted, nil) if merge != tt.wantMerge { var total uint64 @@ -294,7 +297,10 @@ func TestRandomWritesBloatDetection(t *testing.T) { } } - compacted, _ := filer.CompactFileChunks(context.Background(), nil, chunks) + compacted, _, err := filer.CompactFileChunks(context.Background(), nil, chunks) + if err != nil { + t.Fatal(err) + } totalCompacted, reportedFileSize, merge := shouldMergeChunks(compacted, nil) // Sanity: condition must be consistent with its own inputs. @@ -337,7 +343,10 @@ func TestFlushCycleManifestAccumulation(t *testing.T) { // --- flush pipeline (mirrors flushMetadataToFiler) --- manifestChunks, nonManifestChunks := filer.SeparateManifestChunks(entryChunks) - compacted, _ := filer.CompactFileChunks(context.Background(), nil, nonManifestChunks) + compacted, _, err := filer.CompactFileChunks(context.Background(), nil, nonManifestChunks) + if err != nil { + t.Fatal(err) + } _, _, merge := shouldMergeChunks(compacted, manifestChunks) if merge { @@ -406,9 +415,18 @@ func TestVisibleContentPreservedAfterCompact(t *testing.T) { }) } - 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) + origViews, err := filer.ViewFromChunks(context.Background(), nil, chunks, 0, math.MaxInt64) + if err != nil { + t.Fatal(err) + } + compacted, _, err := filer.CompactFileChunks(context.Background(), nil, chunks) + if err != nil { + t.Fatal(err) + } + compViews, err := filer.ViewFromChunks(context.Background(), nil, compacted, 0, math.MaxInt64) + if err != nil { + t.Fatal(err) + } // Collect all (offset, size) pairs from views. type viewKey struct { diff --git a/weed/mount/weedfs_metadata_flush.go b/weed/mount/weedfs_metadata_flush.go index 2a4f3eee9..bce2e9e8d 100644 --- a/weed/mount/weedfs_metadata_flush.go +++ b/weed/mount/weedfs_metadata_flush.go @@ -2,6 +2,7 @@ package mount import ( "context" + "fmt" "sync" "time" @@ -136,7 +137,10 @@ func (wfs *WFS) flushFileMetadata(fh *FileHandle) error { manifestChunks, nonManifestChunks := filer.SeparateManifestChunks(snapshotChunks) // Compact chunks to remove fully overlapped ones - compactedChunks, _ := filer.CompactFileChunks(context.Background(), wfs.LookupFn(), nonManifestChunks) + compactedChunks, _, err := filer.CompactFileChunks(context.Background(), wfs.LookupFn(), nonManifestChunks) + if err != nil { + return fmt.Errorf("compact chunks of %s: %w", fileFullPath, err) + } // Try to create manifest chunks for large files // no chunk deleter here: a failed fold reports the blobs it saved diff --git a/weed/mq/logstore/read_parquet_to_log.go b/weed/mq/logstore/read_parquet_to_log.go index 87366d464..68d654112 100644 --- a/weed/mq/logstore/read_parquet_to_log.go +++ b/weed/mq/logstore/read_parquet_to_log.go @@ -98,7 +98,10 @@ func GenParquetReadFunc(filerClient filer_pb.FilerClient, t topic.Topic, p topic eachFileFn := func(entry *filer_pb.Entry, eachLogEntryFn log_buffer.EachLogEntryFuncType, starTsNs, stopTsNs int64) (processedTsNs int64, err error) { // create readerAt for the parquet file fileSize := filer.FileSize(entry) - visibleIntervals, _ := filer.NonOverlappingVisibleIntervals(context.Background(), lookupFileIdFn, entry.Chunks, 0, int64(fileSize)) + visibleIntervals, err := filer.NonOverlappingVisibleIntervals(context.Background(), lookupFileIdFn, entry.Chunks, 0, int64(fileSize)) + if err != nil { + return 0, fmt.Errorf("resolve chunks of %s: %w", entry.Name, err) + } chunkViews := filer.ViewFromVisibleIntervals(visibleIntervals, 0, int64(fileSize)) readerCache := filer.NewReaderCache(32, chunkCache, lookupFileIdFn, nil) readerAt := filer.NewChunkReaderAtFromClient(context.Background(), readerCache, chunkViews, int64(fileSize), filer.DefaultPrefetchCount) diff --git a/weed/query/engine/aggregations.go b/weed/query/engine/aggregations.go index 54130212e..e672e6174 100644 --- a/weed/query/engine/aggregations.go +++ b/weed/query/engine/aggregations.go @@ -116,8 +116,9 @@ func (opt *FastPathOptimizer) CollectDataSourcesWithTimeFilter(ctx context.Conte // Read parquet file statistics parquetStats, err := hybridScanner.ReadParquetStatistics(partitionPath) if err != nil { - if isDebugMode(ctx) { - fmt.Printf(" ERROR: Failed to read parquet statistics: %v\n", err) + return dataSources, DataSourceError{ + Source: "parquet_statistics", + Cause: fmt.Errorf("partition %s: %w", partitionPath, err), } } else if len(parquetStats) == 0 { if isDebugMode(ctx) { @@ -141,8 +142,9 @@ func (opt *FastPathOptimizer) CollectDataSourcesWithTimeFilter(ctx context.Conte parquetSources := opt.engine.extractParquetSourceFiles(dataSources.ParquetFiles[partitionPath]) liveLogCount, liveLogErr := opt.engine.countLiveLogRowsExcludingParquetSources(ctx, partitionPath, parquetSources) if liveLogErr != nil { - if isDebugMode(ctx) { - fmt.Printf(" ERROR: Failed to count live log rows: %v\n", liveLogErr) + return dataSources, DataSourceError{ + Source: "live_log_count", + Cause: fmt.Errorf("partition %s: %w", partitionPath, liveLogErr), } } else { dataSources.LiveLogRowCount += liveLogCount @@ -308,7 +310,7 @@ func (comp *AggregationComputer) computeGlobalMin(spec AggregationSpec, dataSour liveLogMin, _, err := comp.engine.computeLiveLogMinMax(partition, spec.Column, partitionParquetSources) if err != nil { - continue // Skip partitions with errors + return nil, err } if liveLogMin != nil { @@ -384,7 +386,7 @@ func (comp *AggregationComputer) computeGlobalMax(spec AggregationSpec, dataSour _, liveLogMax, err := comp.engine.computeLiveLogMinMax(partition, spec.Column, partitionParquetSources) if err != nil { - continue // Skip partitions with errors + return nil, err } if liveLogMax != nil { diff --git a/weed/query/engine/engine.go b/weed/query/engine/engine.go index 7a1c783ba..ec8707c0c 100644 --- a/weed/query/engine/engine.go +++ b/weed/query/engine/engine.go @@ -3894,8 +3894,7 @@ func (e *SQLEngine) computeLiveLogMinMax(partitionPath string, columnName string // Scan this log file for MIN/MAX values fileMin, fileMax, err := e.computeFileMinMax(filerClient, filePath, columnName) if err != nil { - fmt.Printf("Warning: failed to compute min/max for file %s: %v\n", filePath, err) - return nil // Continue with other files + return err } // Update global min/max @@ -4030,7 +4029,10 @@ func (e *SQLEngine) eachLogEntryInFile(filerClient filer_pb.FilerClient, filePat // Read file chunks and process them (pattern from countRowsInLogFile) fileSize := filer.FileSize(fileEntry) - visibleIntervals, _ := filer.NonOverlappingVisibleIntervals(context.Background(), lookupFileIdFn, fileEntry.Chunks, 0, int64(fileSize)) + visibleIntervals, err := filer.NonOverlappingVisibleIntervals(context.Background(), lookupFileIdFn, fileEntry.Chunks, 0, int64(fileSize)) + if err != nil { + return err + } chunkViews := filer.ViewFromVisibleIntervals(visibleIntervals, 0, int64(fileSize)) for x := chunkViews.Front(); x != nil; x = x.Next { @@ -4185,8 +4187,7 @@ func (e *SQLEngine) countLiveLogRowsExcludingParquetSources(ctx context.Context, // Count rows in live log file rowCount, err := e.countRowsInLogFile(filerClient, partitionPath, entry) if err != nil { - fmt.Printf("Warning: failed to count rows in %s/%s: %v\n", partitionPath, entry.Name, err) - return nil // Continue with other files + return err } totalRows += rowCount return nil @@ -4354,7 +4355,10 @@ func (e *SQLEngine) countRowsInLogFile(filerClient filer_pb.FilerClient, partiti // Read file chunks and process them (pattern from read_log_from_disk.go) fileSize := filer.FileSize(entry) - visibleIntervals, _ := filer.NonOverlappingVisibleIntervals(context.Background(), lookupFileIdFn, entry.Chunks, 0, int64(fileSize)) + visibleIntervals, err := filer.NonOverlappingVisibleIntervals(context.Background(), lookupFileIdFn, entry.Chunks, 0, int64(fileSize)) + if err != nil { + return 0, err + } chunkViews := filer.ViewFromVisibleIntervals(visibleIntervals, 0, int64(fileSize)) for x := chunkViews.Front(); x != nil; x = x.Next { @@ -4456,23 +4460,22 @@ func (e *SQLEngine) getTopicTotalRowCount(ctx context.Context, namespace, topicN // For each partition, count both parquet and live log rows for _, partition := range partitions { // Count parquet rows - parquetStats, parquetErr := hybridScanner.ReadParquetStatistics(partition) - if parquetErr == nil { - for _, stats := range parquetStats { - totalRowCount += stats.RowCount - } + parquetStats, err := hybridScanner.ReadParquetStatistics(partition) + if err != nil { + return 0, fmt.Errorf("read parquet statistics for %s: %w", partition, err) + } + for _, stats := range parquetStats { + totalRowCount += stats.RowCount } // Count live log rows (with deduplication) - parquetSourceFiles := make(map[string]bool) - if parquetErr == nil { - parquetSourceFiles = e.extractParquetSourceFiles(parquetStats) - } + parquetSourceFiles := e.extractParquetSourceFiles(parquetStats) - liveLogCount, liveLogErr := e.countLiveLogRowsExcludingParquetSources(ctx, partition, parquetSourceFiles) - if liveLogErr == nil { - totalRowCount += liveLogCount + liveLogCount, err := e.countLiveLogRowsExcludingParquetSources(ctx, partition, parquetSourceFiles) + if err != nil { + return 0, fmt.Errorf("count live log rows for %s: %w", partition, err) } + totalRowCount += liveLogCount } return totalRowCount, nil @@ -4509,17 +4512,18 @@ func (e *SQLEngine) getActualRowsScannedForFastPath(ctx context.Context, namespa // (parquet files use metadata/statistics, so they contribute 0 to scan count) for _, partition := range partitions { // Get parquet files to determine what was converted - parquetStats, parquetErr := hybridScanner.ReadParquetStatistics(partition) - parquetSourceFiles := make(map[string]bool) - if parquetErr == nil { - parquetSourceFiles = e.extractParquetSourceFiles(parquetStats) + parquetStats, err := hybridScanner.ReadParquetStatistics(partition) + if err != nil { + return 0, fmt.Errorf("read parquet statistics for %s: %w", partition, err) } + parquetSourceFiles := e.extractParquetSourceFiles(parquetStats) // Count only live log rows that haven't been converted to parquet - liveLogCount, liveLogErr := e.countLiveLogRowsExcludingParquetSources(ctx, partition, parquetSourceFiles) - if liveLogErr == nil { - totalScannedRows += liveLogCount + liveLogCount, err := e.countLiveLogRowsExcludingParquetSources(ctx, partition, parquetSourceFiles) + if err != nil { + return 0, fmt.Errorf("count live log rows for %s: %w", partition, err) } + totalScannedRows += liveLogCount // Note: Parquet files contribute 0 to scan count since we use their metadata/statistics } diff --git a/weed/query/engine/hybrid_message_scanner.go b/weed/query/engine/hybrid_message_scanner.go index 2b45e9cdc..e7836ee8c 100644 --- a/weed/query/engine/hybrid_message_scanner.go +++ b/weed/query/engine/hybrid_message_scanner.go @@ -1140,9 +1140,7 @@ func (h *HybridMessageScanner) ReadParquetStatistics(partitionPath string) ([]*P // Extract statistics from this parquet file stats, err := h.extractParquetFileStats(entry, lookupFileIdFn, chunkCache) if err != nil { - // Log error but continue processing other files - fmt.Printf("Warning: failed to extract stats from %s: %v\n", entry.Name, err) - return nil + return err } if stats != nil { @@ -1164,7 +1162,10 @@ func (h *HybridMessageScanner) extractParquetFileStats(entry *filer_pb.Entry, lo } // Create reader for the parquet file fileSize := filer.FileSize(entry) - visibleIntervals, _ := filer.NonOverlappingVisibleIntervals(context.Background(), lookupFileIdFn, entry.Chunks, 0, int64(fileSize)) + visibleIntervals, err := filer.NonOverlappingVisibleIntervals(context.Background(), lookupFileIdFn, entry.Chunks, 0, int64(fileSize)) + if err != nil { + return nil, fmt.Errorf("resolve chunks of %s: %w", entry.Name, err) + } chunkViews := filer.ViewFromVisibleIntervals(visibleIntervals, 0, int64(fileSize)) readerCache := filer.NewReaderCache(32, chunkCache, lookupFileIdFn, nil) readerAt := filer.NewChunkReaderAtFromClient(context.Background(), readerCache, chunkViews, int64(fileSize), filer.DefaultPrefetchCount) diff --git a/weed/replication/sink/azuresink/azure_sink.go b/weed/replication/sink/azuresink/azure_sink.go index 43374de44..1ce8d62cd 100644 --- a/weed/replication/sink/azuresink/azure_sink.go +++ b/weed/replication/sink/azuresink/azure_sink.go @@ -130,7 +130,10 @@ func (g *AzureSink) CreateEntry(key string, entry *filer_pb.Entry, signatures [] } totalSize := filer.FileSize(entry) - chunkViews := filer.ViewFromChunks(context.Background(), g.filerSource.LookupFileId, entry.GetChunks(), 0, int64(totalSize)) + chunkViews, err := filer.ViewFromChunks(context.Background(), g.filerSource.LookupFileId, entry.GetChunks(), 0, int64(totalSize)) + if err != nil { + return fmt.Errorf("resolve chunks of %s: %w", key, err) + } // Create append blob client appendBlobClient := g.client.ServiceClient().NewContainerClient(g.container).NewAppendBlobClient(key) @@ -138,7 +141,7 @@ func (g *AzureSink) CreateEntry(key string, entry *filer_pb.Entry, signatures [] // Try to create the blob first (without access conditions for initial creation) ctxCreate, cancelCreate := context.WithTimeout(context.Background(), azure.DefaultAzureOpTimeout) defer cancelCreate() - _, err := appendBlobClient.Create(ctxCreate, nil) + _, err = appendBlobClient.Create(ctxCreate, nil) needsWrite := true freshlyCreated := false diff --git a/weed/replication/sink/b2sink/b2_sink.go b/weed/replication/sink/b2sink/b2_sink.go index a18af0e4f..a3b59a7cf 100644 --- a/weed/replication/sink/b2sink/b2_sink.go +++ b/weed/replication/sink/b2sink/b2_sink.go @@ -104,7 +104,10 @@ func (g *B2Sink) CreateEntry(key string, entry *filer_pb.Entry, signatures []int } totalSize := filer.FileSize(entry) - chunkViews := filer.ViewFromChunks(context.Background(), g.filerSource.LookupFileId, entry.GetChunks(), 0, int64(totalSize)) + chunkViews, err := filer.ViewFromChunks(context.Background(), g.filerSource.LookupFileId, entry.GetChunks(), 0, int64(totalSize)) + if err != nil { + return fmt.Errorf("resolve chunks of %s: %w", key, err) + } bucket, err := g.client.Bucket(context.Background(), g.bucket) if err != nil { diff --git a/weed/replication/sink/gcssink/gcs_sink.go b/weed/replication/sink/gcssink/gcs_sink.go index f2aa271b5..ff2879add 100644 --- a/weed/replication/sink/gcssink/gcs_sink.go +++ b/weed/replication/sink/gcssink/gcs_sink.go @@ -117,7 +117,10 @@ func (g *GcsSink) CreateEntry(key string, entry *filer_pb.Entry, signatures []in } totalSize := filer.FileSize(entry) - chunkViews := filer.ViewFromChunks(context.Background(), g.filerSource.LookupFileId, entry.GetChunks(), 0, int64(totalSize)) + chunkViews, err := filer.ViewFromChunks(context.Background(), g.filerSource.LookupFileId, entry.GetChunks(), 0, int64(totalSize)) + if err != nil { + return fmt.Errorf("resolve chunks of %s: %w", key, err) + } ctx, cancel := context.WithCancel(context.Background()) defer cancel() diff --git a/weed/replication/sink/localsink/local_sink.go b/weed/replication/sink/localsink/local_sink.go index f316b80ed..8debf7993 100644 --- a/weed/replication/sink/localsink/local_sink.go +++ b/weed/replication/sink/localsink/local_sink.go @@ -84,7 +84,10 @@ func (localsink *LocalSink) CreateEntry(key string, entry *filer_pb.Entry, signa glog.V(4).Infof("Create Entry key: %s", key) totalSize := filer.FileSize(entry) - chunkViews := filer.ViewFromChunks(context.Background(), localsink.filerSource.LookupFileId, entry.GetChunks(), 0, int64(totalSize)) + chunkViews, err := filer.ViewFromChunks(context.Background(), localsink.filerSource.LookupFileId, entry.GetChunks(), 0, int64(totalSize)) + if err != nil { + return fmt.Errorf("resolve chunks of %s: %w", key, err) + } dir := filepath.Dir(key) diff --git a/weed/s3api/s3api_object_handlers.go b/weed/s3api/s3api_object_handlers.go index a99c93b1c..0065eb863 100644 --- a/weed/s3api/s3api_object_handlers.go +++ b/weed/s3api/s3api_object_handlers.go @@ -1807,7 +1807,10 @@ func (s3a *S3ApiServer) streamDecryptedRangeFromChunks(ctx context.Context, w io if err != nil { return 0, err } - chunkViews := filer.ViewFromChunks(ctx, nil, resolvedChunks, offset, size) + chunkViews, err := filer.ViewFromChunks(ctx, nil, resolvedChunks, offset, size) + if err != nil { + return 0, err + } totalWritten := int64(0) targetOffset := offset diff --git a/weed/server/filer_grpc_server.go b/weed/server/filer_grpc_server.go index 8759afbaa..862b9b4fa 100644 --- a/weed/server/filer_grpc_server.go +++ b/weed/server/filer_grpc_server.go @@ -803,7 +803,10 @@ func (fs *FilerServer) cleanupChunks(ctx context.Context, fullpath string, exist // files with manifest chunks are usually large and append only, skip calculating covered chunks manifestChunks, nonManifestChunks := filer.SeparateManifestChunks(newEntry.GetChunks()) - chunks, coveredChunks := filer.CompactFileChunks(ctx, fs.lookupFileId, nonManifestChunks) + chunks, coveredChunks, err := filer.CompactFileChunks(ctx, fs.lookupFileId, nonManifestChunks) + if err != nil { + return newEntry.GetChunks(), nil, fmt.Errorf("CompactFileChunks: %w", err) + } garbage = append(garbage, coveredChunks...) if newEntry.Attributes != nil { diff --git a/weed/server/filer_grpc_server_cleanup_chunks_test.go b/weed/server/filer_grpc_server_cleanup_chunks_test.go new file mode 100644 index 000000000..7aeef26fd --- /dev/null +++ b/weed/server/filer_grpc_server_cleanup_chunks_test.go @@ -0,0 +1,47 @@ +package weed_server + +import ( + "context" + "strings" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +// A CreateEntry whose context is cancelled before its chunks are compacted must +// fail the request and store nothing, rather than panic in CompactFileChunks. +func TestCreateEntryCancelledContextFailsRequest(t *testing.T) { + store := newRenameTestStore() + f := newRenameTestFiler(t, store) + + fs := &FilerServer{ + filer: f, + option: &FilerOption{}, + entryLockTable: util.NewLockTable[util.FullPath](), + } + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + _, err := fs.CreateEntry(ctx, &filer_pb.CreateEntryRequest{ + Directory: "/test", + SkipCheckParentDirectory: true, + Entry: &filer_pb.Entry{ + Name: "obj", + Attributes: &filer_pb.FuseAttributes{Mtime: 1700000000, FileMode: 0644, Inode: 1}, + Chunks: []*filer_pb.FileChunk{ + {FileId: "1,01", Offset: 0, Size: 100, ModifiedTsNs: 1}, + }, + }, + }) + if err == nil || !strings.Contains(err.Error(), "CompactFileChunks: "+context.Canceled.Error()) { + t.Fatalf("err = %v, want CompactFileChunks to fail with context canceled", err) + } + + store.mu.Lock() + defer store.mu.Unlock() + if _, found := store.entries["/test/obj"]; found { + t.Fatal("entry was stored despite the failed request") + } +}