From 5553e7b876c5db87f54e3e2f26efd36ebe39cff6 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Wed, 15 Jul 2026 22:01:19 -0700 Subject: [PATCH] mount: surface ENOSPC instead of endless waiting when the cluster is full (#10341) When every volume is full, a chunk upload fails only after the assign retry budget, the failed chunk's data is dropped, and the mount kept accepting writes anyway. cp would crawl for hours pushing the rest of the file through a pipeline that could not persist it, and only close() reported an error - a generic EIO. Poison the file handle on the first failed chunk upload so subsequent writes fail immediately, and map "no writable volumes" / "no free volumes" upload errors to ENOSPC so the writing process aborts with "No space left on device". Also guard lastErr with a mutex: it was written concurrently by uploader goroutines, and keep the first error so later failures do not mask the root cause. --- weed/mount/dirty_pages_chunked.go | 29 ++++++++++++++++++--- weed/mount/dirty_pages_chunked_test.go | 31 +++++++++++++++++++++++ weed/mount/error_classifier.go | 22 ++++++++++++++++ weed/mount/error_classifier_test.go | 35 ++++++++++++++++++++++++++ weed/mount/weedfs_file_copy_range.go | 2 +- weed/mount/weedfs_file_sync.go | 2 +- weed/mount/weedfs_file_write.go | 2 +- 7 files changed, 117 insertions(+), 6 deletions(-) create mode 100644 weed/mount/dirty_pages_chunked_test.go diff --git a/weed/mount/dirty_pages_chunked.go b/weed/mount/dirty_pages_chunked.go index 4cd21d1a5..29ff24b3e 100644 --- a/weed/mount/dirty_pages_chunked.go +++ b/weed/mount/dirty_pages_chunked.go @@ -13,6 +13,7 @@ import ( type ChunkedDirtyPages struct { fh *FileHandle writeWaitGroup sync.WaitGroup + lastErrLock sync.Mutex lastErr error collection string replication string @@ -40,6 +41,12 @@ func newMemoryChunkPages(fh *FileHandle, chunkSize int64) *ChunkedDirtyPages { } func (pages *ChunkedDirtyPages) AddPage(offset int64, data []byte, isSequential bool, tsNs int64) error { + // A failed chunk upload already dropped its data, so this handle can never + // flush cleanly. Reject further writes so the writer fails fast instead of + // pushing the rest of the file through a pipeline that cannot persist it. + if err := pages.LastError(); err != nil { + return err + } pages.hasWrites = true glog.V(4).Infof("%v memory AddPage [%d, %d)", pages.fh.fh, offset, offset+int64(len(data))) @@ -53,12 +60,28 @@ func (pages *ChunkedDirtyPages) FlushData() error { return nil } pages.uploadPipeline.FlushAll() - if pages.lastErr != nil { - return fmt.Errorf("flush data: %v", pages.lastErr) + if err := pages.LastError(); err != nil { + return fmt.Errorf("flush data: %w", err) } return nil } +// LastError returns the first chunk upload failure on this handle. It is +// sticky: the failed chunk's data is gone, so the handle stays poisoned. +func (pages *ChunkedDirtyPages) LastError() error { + pages.lastErrLock.Lock() + defer pages.lastErrLock.Unlock() + return pages.lastErr +} + +func (pages *ChunkedDirtyPages) setLastError(err error) { + pages.lastErrLock.Lock() + defer pages.lastErrLock.Unlock() + if pages.lastErr == nil { + pages.lastErr = err + } +} + func (pages *ChunkedDirtyPages) ReadDirtyDataAt(data []byte, startOffset int64, tsNs int64) (maxStop int64) { if !pages.hasWrites { return @@ -75,7 +98,7 @@ func (pages *ChunkedDirtyPages) saveChunkedFileIntervalToStorage(reader io.Reade chunk, err := pages.fh.wfs.saveDataAsChunk(fileFullPath)(reader, fileName, offset, modifiedTsNs, uint64(size)) if err != nil { glog.V(0).Infof("%v saveToStorage [%d,%d): %v", fileFullPath, offset, offset+size, err) - pages.lastErr = err + pages.setLastError(err) return } pages.fh.AddChunks([]*filer_pb.FileChunk{chunk}) diff --git a/weed/mount/dirty_pages_chunked_test.go b/weed/mount/dirty_pages_chunked_test.go new file mode 100644 index 000000000..7028f2953 --- /dev/null +++ b/weed/mount/dirty_pages_chunked_test.go @@ -0,0 +1,31 @@ +package mount + +import ( + "errors" + "testing" +) + +// Once a chunk upload fails its data is gone, so the handle must reject +// further writes instead of buffering the rest of the file — against a full +// cluster that turned cp into an hours-long crawl that only errored at close. +func TestChunkedDirtyPagesFailWritesAfterUploadError(t *testing.T) { + fh := &FileHandle{wfs: &WFS{option: &Option{}}} + pages := newMemoryChunkPages(fh, 1024) + defer pages.Destroy() + pages.hasWrites = true + + uploadErr := errors.New("assign volume failure: no writable volumes") + pages.setLastError(uploadErr) + + if err := pages.AddPage(0, []byte("x"), true, 1); !errors.Is(err, uploadErr) { + t.Fatalf("AddPage after upload failure = %v, want sticky %v", err, uploadErr) + } + if err := pages.FlushData(); !errors.Is(err, uploadErr) { + t.Fatalf("FlushData after upload failure = %v, want wrapped %v", err, uploadErr) + } + // First failure wins; later errors must not mask the root cause. + pages.setLastError(errors.New("later error")) + if err := pages.LastError(); !errors.Is(err, uploadErr) { + t.Fatalf("LastError = %v, want first error %v", err, uploadErr) + } +} diff --git a/weed/mount/error_classifier.go b/weed/mount/error_classifier.go index 6b55a610f..22e664d32 100644 --- a/weed/mount/error_classifier.go +++ b/weed/mount/error_classifier.go @@ -48,6 +48,28 @@ func grpcErrorToFuseStatus(err error) fuse.Status { return fuse.EIO } +// isStorageFullError reports whether a chunk upload failed because the cluster +// has no writable volume left (disks full and volume growth failed). The +// master's message reaches the mount as plain text inside +// AssignVolumeResponse.Error, so match on text rather than a status code. +func isStorageFullError(err error) bool { + if err == nil { + return false + } + errStr := strings.ToLower(err.Error()) + return strings.Contains(errStr, "no writable volumes") || + strings.Contains(errStr, "no free volumes") +} + +// writeErrorToFuseStatus maps a dirty-page write/flush failure to the status +// returned to the kernel: ENOSPC when the cluster is out of space, EIO otherwise. +func writeErrorToFuseStatus(err error) fuse.Status { + if isStorageFullError(err) { + return fuse.Status(syscall.ENOSPC) + } + return fuse.EIO +} + // isRetryableFilerError reports whether a filer RPC error looks transient // enough to retry. It takes a conservative whitelist approach: only errors // that clearly describe a permanent application-level failure diff --git a/weed/mount/error_classifier_test.go b/weed/mount/error_classifier_test.go index b49db4f99..04cb44cab 100644 --- a/weed/mount/error_classifier_test.go +++ b/weed/mount/error_classifier_test.go @@ -44,6 +44,41 @@ func TestGrpcErrorToFuseStatusDropsCanceledThroughPercentV(t *testing.T) { } } +// A full cluster must surface as ENOSPC so cp/rsync abort with "No space left +// on device" instead of a generic I/O error. The master's message reaches the +// mount stringified inside AssignVolumeResponse.Error, in one of two shapes: +// the load-shedding gRPC status text or the topology's plain error text. +func TestWriteErrorToFuseStatus(t *testing.T) { + cases := []struct { + name string + err error + want fuse.Status + }{ + { + "assign shed during growth", + errors.New("upload data: filerGrpcAddress assign volume: assign volume failure count:1: assign volume: rpc error: code = ResourceExhausted desc = no writable volumes for replication:000 collection:, volume growth in progress"), + fuse.Status(syscall.ENOSPC), + }, + { + "no free volumes left", + errors.New("flush data: upload data: assign volume: No writable volumes available for collection: replication:000 ttl: and no free volumes left for {replication:000}"), + fuse.Status(syscall.ENOSPC), + }, + { + "unrelated upload failure", + errors.New("upload data: put http://volume:8080/3,0144: connection reset by peer"), + fuse.EIO, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := writeErrorToFuseStatus(tc.err); got != tc.want { + t.Fatalf("writeErrorToFuseStatus(%v) = %v, want %v", tc.err, got, tc.want) + } + }) + } +} + func TestIsRetryableFilerError(t *testing.T) { cases := []struct { name string diff --git a/weed/mount/weedfs_file_copy_range.go b/weed/mount/weedfs_file_copy_range.go index da5fd7f50..eb9ca4282 100644 --- a/weed/mount/weedfs_file_copy_range.go +++ b/weed/mount/weedfs_file_copy_range.go @@ -186,7 +186,7 @@ func (wfs *WFS) CopyFileRange(cancel <-chan struct{}, in *fuse.CopyFileRangeIn) fhOut.dirtyPages.writerPattern.IsSequentialMode(), nowUnixNano); err != nil { glog.Errorf("AddPage error: %v", err) - return 0, fuse.EIO + return 0, writeErrorToFuseStatus(err) } // Accumulate for the next loop iteration diff --git a/weed/mount/weedfs_file_sync.go b/weed/mount/weedfs_file_sync.go index 487e6d03d..57712d8a8 100644 --- a/weed/mount/weedfs_file_sync.go +++ b/weed/mount/weedfs_file_sync.go @@ -161,7 +161,7 @@ func (wfs *WFS) doFlush(ctx context.Context, fh *FileHandle, uid, gid uint32, al if !isOverQuota { if err := fh.dirtyPages.FlushData(); err != nil { glog.Errorf("%v doFlush: %v", fileFullPath, err) - return fuse.EIO + return writeErrorToFuseStatus(err) } } diff --git a/weed/mount/weedfs_file_write.go b/weed/mount/weedfs_file_write.go index a70932ff1..9c10a0df6 100644 --- a/weed/mount/weedfs_file_write.go +++ b/weed/mount/weedfs_file_write.go @@ -85,7 +85,7 @@ func (wfs *WFS) Write(cancel <-chan struct{}, in *fuse.WriteIn, data []byte) (wr if err := fh.dirtyPages.AddPage(offset, data, fh.dirtyPages.writerPattern.IsSequentialMode(), tsNs); err != nil { glog.Errorf("AddPage error: %v", err) - return 0, fuse.EIO + return 0, writeErrorToFuseStatus(err) } written = uint32(len(data))