local: make out of space errors fatal during multi-thread transfers

backend/local/local.go

@@ -1634,12 +1634,17 @@ func isDiskFullError(err error) bool {
 	return errors.Is(err, file.ErrDiskFull) || fserrors.IsErrNoSpace(err)
 }
 
+func wrapFatalIfNoSpace(err error, enabled bool) error {
+	if err != nil && enabled && isDiskFullError(err) {
+		return fserrors.FatalError(err)
+	}
+	return err
+}
+
 // Update the object from in with modTime and size
 func (o *Object) Update(ctx context.Context, in io.Reader, src fs.ObjectInfo, options ...fs.OpenOption) (err error) {
 	defer func() {
-		if err != nil && o.fs.opt.FatalIfNoSpace && isDiskFullError(err) {
+		err = wrapFatalIfNoSpace(err, o.fs.opt.FatalIfNoSpace)
-			err = fserrors.FatalError(err)
-		}
 	}()
 	var out io.WriteCloser
 	var hasher *hash.MultiHasher
@@ -1762,12 +1767,30 @@ func (o *Object) Update(ctx context.Context, in io.Reader, src fs.ObjectInfo, op
 
 var sparseWarning sync.Once
 
+type fatalIfNoSpaceWriterAt struct {
+	fs.WriterAtCloser
+	enabled bool
+}
+
+func (w *fatalIfNoSpaceWriterAt) WriteAt(p []byte, off int64) (n int, err error) {
+	n, err = w.WriterAtCloser.WriteAt(p, off)
+	return n, wrapFatalIfNoSpace(err, w.enabled)
+}
+
+func (w *fatalIfNoSpaceWriterAt) Close() error {
+	return wrapFatalIfNoSpace(w.WriterAtCloser.Close(), w.enabled)
+}
+
 // OpenWriterAt opens with a handle for random access writes
 //
 // Pass in the remote desired and the size if known.
 //
 // It truncates any existing object
-func (f *Fs) OpenWriterAt(ctx context.Context, remote string, size int64) (fs.WriterAtCloser, error) {
+func (f *Fs) OpenWriterAt(ctx context.Context, remote string, size int64) (writer fs.WriterAtCloser, err error) {
+	defer func() {
+		err = wrapFatalIfNoSpace(err, f.opt.FatalIfNoSpace)
+	}()
+
 	// Temporary Object under construction
 	o, err := f.newObject(remote)
 	if err != nil {
@@ -1805,7 +1828,10 @@ func (f *Fs) OpenWriterAt(ctx context.Context, remote string, size int64) (fs.Wr
 		}
 	}
 
-	return out, nil
+	return &fatalIfNoSpaceWriterAt{
+		WriterAtCloser: out,
+		enabled:        f.opt.FatalIfNoSpace,
+	}, nil
 }
 
 // setMetadata sets the file info from the os.FileInfo passed in

backend/local/local_internal_diskfull_test.go

@@ -60,7 +60,7 @@ func TestIsDiskFullError(t *testing.T) {
 // FatalIfNoSpace setting, returning the error from Update.
 func updateWithReader(t *testing.T, fatalIfNoSpace bool, readerErr error) error {
 	t.Helper()
-	r := fstest.NewRun(t)
+	r := fstest.NewRunIndividual(t)
 	f := r.Flocal.(*Fs)
 	f.opt.FatalIfNoSpace = fatalIfNoSpace
 
@@ -96,3 +96,87 @@ func TestUpdateFatalIfNoSpaceOnButNotDiskFull(t *testing.T) {
 	require.Error(t, err)
 	assert.False(t, fserrors.IsFatalError(err), "non-disk-full errors must not be fatal regardless of option")
 }
+
+type diskFullWriterAtCloser struct {
+	writeErr error
+	closeErr error
+	n        int
+	data     []byte
+	offset   int64
+	closed   bool
+}
+
+func (w *diskFullWriterAtCloser) WriteAt(p []byte, off int64) (int, error) {
+	w.data = append([]byte(nil), p...)
+	w.offset = off
+	return w.n, w.writeErr
+}
+
+func (w *diskFullWriterAtCloser) Close() error {
+	w.closed = true
+	return w.closeErr
+}
+
+func testOpenWriterAtError(t *testing.T, injected error, diskFull bool) {
+	t.Helper()
+	for _, enabled := range []bool{false, true} {
+		t.Run(fmt.Sprintf("enabled=%v", enabled), func(t *testing.T) {
+			r := fstest.NewRunIndividual(t)
+			f := r.Flocal.(*Fs)
+			f.opt.FatalIfNoSpace = enabled
+			f.opt.NoPreAllocate = true
+			writer, err := f.OpenWriterAt(context.Background(), "test.txt", 0)
+			require.NoError(t, err)
+			wrapped, ok := writer.(*fatalIfNoSpaceWriterAt)
+			require.True(t, ok)
+			require.NoError(t, wrapped.WriterAtCloser.Close())
+			wantN := 2
+			if injected == nil {
+				wantN = 4
+			}
+			underlying := &diskFullWriterAtCloser{writeErr: injected, closeErr: injected, n: wantN}
+			wrapped.WriterAtCloser = underlying
+			n, err := writer.WriteAt([]byte("data"), 17)
+			assert.Equal(t, wantN, n)
+			assert.Equal(t, []byte("data"), underlying.data)
+			assert.Equal(t, int64(17), underlying.offset)
+			assert.ErrorIs(t, err, injected)
+			assert.Equal(t, enabled && diskFull, fserrors.IsFatalError(err))
+			err = writer.Close()
+			assert.True(t, underlying.closed)
+			assert.ErrorIs(t, err, injected)
+			assert.Equal(t, enabled && diskFull, fserrors.IsFatalError(err))
+		})
+	}
+}
+
+func TestOpenWriterAtFatalIfNoSpace(t *testing.T) {
+	for _, test := range []struct {
+		name     string
+		err      error
+		diskFull bool
+	}{
+		{"success", nil, false},
+		{"unrelated", syscall.EPERM, false},
+		{"ENOSPC", syscall.ENOSPC, true},
+		{"ErrDiskFull", file.ErrDiskFull, true},
+		{"wrapped", fmt.Errorf("write: %w", syscall.ENOSPC), true},
+	} {
+		t.Run(test.name, func(t *testing.T) { testOpenWriterAtError(t, test.err, test.diskFull) })
+	}
+}
+
+func TestOpenWriterAtSetupError(t *testing.T) {
+	for _, enabled := range []bool{false, true} {
+		t.Run(fmt.Sprintf("enabled=%v", enabled), func(t *testing.T) {
+			r := fstest.NewRunIndividual(t)
+			f := r.Flocal.(*Fs)
+			f.opt.FatalIfNoSpace = enabled
+			require.NoError(t, os.WriteFile(filepath.Join(r.LocalName, "parent"), nil, 0600))
+			writer, err := f.OpenWriterAt(context.Background(), "parent/child", 0)
+			require.Error(t, err)
+			assert.Nil(t, writer)
+			assert.False(t, fserrors.IsFatalError(err))
+		})
+	}
+}

backend/local/local_internal_diskfull_windows_test.go

@@ -35,3 +35,18 @@ func TestUpdateFatalIfNoSpaceWindows(t *testing.T) {
 		})
 	}
 }
+
+func TestOpenWriterAtFatalIfNoSpaceWindows(t *testing.T) {
+	for _, test := range []struct {
+		name     string
+		err      error
+		diskFull bool
+	}{
+		{"ERROR_DISK_FULL", windows.ERROR_DISK_FULL, true},
+		{"ERROR_HANDLE_DISK_FULL", windows.ERROR_HANDLE_DISK_FULL, true},
+		{"PathError", &os.PathError{Op: "write", Path: "test.txt", Err: windows.ERROR_DISK_FULL}, true},
+		{"ERROR_ACCESS_DENIED", windows.ERROR_ACCESS_DENIED, false},
+	} {
+		t.Run(test.name, func(t *testing.T) { testOpenWriterAtError(t, test.err, test.diskFull) })
+	}
+}

fs/fserrors/enospc_error_windows_test.go

@@ -4,37 +4,13 @@ package fserrors
 
 import (
 	"os"
-	"path/filepath"
 	"syscall"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
 	"golang.org/x/sys/windows"
 )
 
-func TestIsErrNoSpaceRealWindowsError(t *testing.T) {
-	dir := t.TempDir()
-	dirp, err := windows.UTF16PtrFromString(dir)
-	if err != nil {
-		t.Skipf("cannot convert the temporary directory path: %v", err)
-	}
-	var available, total, free uint64
-	if err := windows.GetDiskFreeSpaceEx(dirp, &available, &total, &free); err != nil {
-		t.Skipf("cannot read the free space of the temporary directory: %v", err)
-	}
-
-	f, err := os.Create(filepath.Join(dir, "truncate"))
-	require.NoError(t, err)
-	defer func() { require.NoError(t, f.Close()) }()
-
-	err = f.Truncate(int64(free + 1<<30))
-	if err == nil {
-		t.Skip("real Windows disk-full error coverage lost: volume did not enforce the free-space limit when truncating the file")
-	}
-	assert.True(t, IsErrNoSpace(err), "error = %v", err)
-}
-
 func TestIsErrNoSpaceWindows(t *testing.T) {
 	tests := []struct {
 		name string

fs/operations/multithread_test.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/rclone/rclone/fs/accounting"
+	"github.com/rclone/rclone/fs/fserrors"
 	"github.com/rclone/rclone/fs/hash"
 	"github.com/rclone/rclone/fs/object"
 	"github.com/rclone/rclone/fstest/mockfs"
@@ -342,3 +343,62 @@ func TestMultithreadCopyAbort(t *testing.T) {
 		require.NoError(t, o.Remove(ctx))
 	}
 }
+
+type errorWriterAtCloser struct {
+	writeErr error
+	closeErr error
+	closed   bool
+}
+
+func (w *errorWriterAtCloser) WriteAt(p []byte, _ int64) (int, error) {
+	if w.writeErr != nil {
+		return 0, w.writeErr
+	}
+	return len(p), nil
+}
+
+func (w *errorWriterAtCloser) Close() error {
+	w.closed = true
+	return w.closeErr
+}
+
+// Classification belongs to the backend; this checks propagation through the consumer.
+func TestMultithreadCopyWriterAtErrors(t *testing.T) {
+	for _, stage := range []string{"open", "write", "close"} {
+		for _, fatal := range []bool{false, true} {
+			t.Run(fmt.Sprintf("%s/fatal=%v", stage, fatal), func(t *testing.T) {
+				ctx, ci := fs.AddConfig(context.Background())
+				ci.MultiThreadChunkSize = 4
+				ci.MultiThreadWriteBufferSize = 0
+				ci.MultiThreadStreams = 2
+				ci.MultiThreadSet = true
+				f, err := mockfs.NewFs(ctx, "destination", "", nil)
+				require.NoError(t, err)
+				src := mockobject.New("file.txt").WithContent([]byte("0123456789abcdef"), mockobject.SeekModeNone)
+				cause := errors.New("disk full")
+				injected := cause
+				if fatal {
+					injected = fserrors.FatalError(injected)
+				}
+				writer := &errorWriterAtCloser{}
+				f.Features().OpenWriterAt = func(context.Context, string, int64) (fs.WriterAtCloser, error) {
+					switch stage {
+					case "open":
+						return nil, injected
+					case "write":
+						writer.writeErr = injected
+					case "close":
+						writer.closeErr = injected
+					}
+					return writer, nil
+				}
+				tr := accounting.GlobalStats().NewTransfer(src, nil)
+				_, err = multiThreadCopy(ctx, f, src.Remote(), src, 2, tr)
+				tr.Done(ctx, err)
+				require.ErrorIs(t, err, cause)
+				assert.Equal(t, fatal, fserrors.IsFatalError(err))
+				assert.Equal(t, stage != "open", writer.closed)
+			})
+		}
+	}
+}

vfs/vfscache/cache_test.go

@@ -674,13 +674,15 @@ func TestCacheKickCleaner(t *testing.T) {
 
 	// Only the cleaner clears the out of space condition, so with no cleaner
 	// running a KickCleaner which waited for it would never return.
-	t.Run("CleanerDisabled", func(t *testing.T) {
+	for _, interval := range []fs.Duration{0, -1} {
-		opt := vfscommon.Opt
+		t.Run("CleanerDisabled/"+interval.String(), func(t *testing.T) {
-		opt.CachePollInterval = 0
+			opt := vfscommon.Opt
-		_, c := newTestCacheOpt(t, opt)
+			opt.CachePollInterval = interval
+			_, c := newTestCacheOpt(t, opt)
+			assert.True(t, kickCleaner(t, c), "KickCleaner did not return with the cleaner disabled")
+		})
+	}
 
-		assert.True(t, kickCleaner(t, c), "KickCleaner did not return with the cleaner disabled")
-	})
 }
 
 func TestCacheSetModTime(t *testing.T) {