vfs: add WriteFileHandle.CloseWithError to abandon streaming writes

Closing a streaming write handle sends a clean EOF to the backend upload, so a writer which knows its data source failed part way through had no way to stop the truncated file being stored as if it were complete. CloseWithError closes the handle failing the upload with the given reason instead. EOF-like reasons are rewritten so the layers reading the pipe can't mistake them for a clean end of stream and commit the partial file. Needed by serve s3 to abort interrupted PUTs - see #9718
2026-08-11 20:58:48 +01:00
parent 7357fb82a9
commit 2f0657c35b
2 changed files with 180 additions and 5 deletions
@@ -1,6 +1,9 @@
 package vfs

 import (
+	"context"
+	"errors"
+	"fmt"
 	"io"
 	"os"
 	"sync"
@@ -21,6 +24,8 @@ type WriteFileHandle struct {
 	o           fs.Object
 	result      chan error
 	file        *File
+	ctx         context.Context    // the streaming upload runs under this
+	cancel      context.CancelFunc // fails the streaming upload, unblocking a stalled write
 	offset      int64
 	flags       int
 	closed      bool // set if handle has been closed
@@ -46,6 +51,7 @@ func newWriteFileHandle(d *Dir, f *File, remote string, flags int) (*WriteFileHa
 		result: make(chan error, 1),
 		file:   f,
 	}
+	fh.ctx, fh.cancel = context.WithCancel(f.ctx)
 	fh.cond = sync.Cond{L: &fh.mu}
 	fh.file.addWriter(fh)
 	return fh, nil
@@ -85,7 +91,7 @@ func (fh *WriteFileHandle) openPending() (err error) {
 		}()
 		defer vfscommon.RecoverPanic(fh.remote, &err)
 		// NB Rcat deals with Stats.Transferring, etc.
-		o, err = operations.Rcat(fh.file.ctx, fh.file.Fs(), fh.remote, pipeReader, time.Now(), nil)
+		o, err = operations.Rcat(fh.ctx, fh.file.Fs(), fh.remote, pipeReader, time.Now(), nil)
 		if err != nil {
 			fs.Errorf(fh.remote, "WriteFileHandle.New Rcat failed: %v", err)
 		}
@@ -196,10 +202,22 @@ func (fh *WriteFileHandle) Offset() (offset int64) {
 //
 // Must be called with fh.mu held
 func (fh *WriteFileHandle) close() (err error) {
+	return fh.closeWithError(nil)
+}
+
+// closeWithError closes the file handle like close does. If reason is
+// non-nil the streaming upload is failed with reason rather than being
+// ended normally, so a partially written file is discarded instead of
+// being stored as if it were complete.
+//
+// Must be called with fh.mu held
+func (fh *WriteFileHandle) closeWithError(reason error) (err error) {
 	if fh.closed {
 		return ECLOSED
 	}
 	fh.closed = true
+	// Release the upload context once the upload has finished
+	defer fh.cancel()
 	// leave writer open until file is transferred
 	defer func() {
 		fh.file.delWriter(fh)
@@ -211,15 +229,22 @@ func (fh *WriteFileHandle) close() (err error) {
 	if err = fh.openPending(); err != nil {
 		return err
 	}
-	writeCloseErr := fh.pipeWriter.Close()
+	writeCloseErr := fh.pipeWriter.CloseWithError(reason)
 	err = <-fh.result
 	if err == nil {
 		fh.file.setObject(fh.o)
 		err = writeCloseErr
-	} else if fh.file.getObject() == nil {
+	} else {
+		if fh.file.getObject() == nil {
 			// Remove vfs file entry when no object is present
 			_ = fh.file.Remove()
 		}
+		if reason != nil {
+			// The upload's own error is usually just the cancellation
+			// the abandon caused - reason is the cause worth reporting.
+			err = reason
+		}
+	}
 	return err
 }

@@ -230,6 +255,34 @@ func (fh *WriteFileHandle) Close() error {
 	return fh.close()
 }

+// CloseWithError closes the file handle, abandoning the write.
+//
+// The streaming upload to the backend is failed with reason rather than
+// being ended normally, so the partially written file is discarded
+// instead of being stored as if it were complete. A stalled upload is
+// interrupted rather than waited for, so this returns promptly even
+// when the backend has stopped accepting data. A nil reason is the
+// same as Close.
+//
+// It returns ECLOSED if the handle has already been closed.
+func (fh *WriteFileHandle) CloseWithError(reason error) error {
+	// Don't let an EOF-like reason be mistaken for a clean end of stream
+	// by whatever is reading the other end of the pipe.
+	if reason != nil && (errors.Is(reason, io.EOF) || errors.Is(reason, io.ErrUnexpectedEOF)) {
+		reason = fmt.Errorf("write aborted: %v", reason)
+	}
+	if reason != nil {
+		// Fail the upload before taking the lock: a write blocked on a
+		// stalled upload holds fh.mu, and only failing the upload
+		// unblocks it - otherwise the abandon would wait out the
+		// backend's timeout (or forever without one).
+		fh.cancel()
+	}
+	fh.mu.Lock()
+	defer fh.mu.Unlock()
+	return fh.closeWithError(reason)
+}
+
 // Flush is called on each close() of a file descriptor. So if a
 // filesystem wants to return write errors in close() and the file has
 // cached dirty data, this is a good place to write back data and
@@ -13,6 +13,7 @@ import (
 	"github.com/rclone/rclone/fs"
 	"github.com/rclone/rclone/fstest"
 	"github.com/rclone/rclone/lib/random"
+	"github.com/rclone/rclone/vfs/vfscommon"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -235,6 +236,47 @@ func TestWriteFileHandleWriteAt(t *testing.T) {
 	fstest.CheckListingWithPrecision(t, r.Fremote, []fstest.Item{file1}, []string{}, fs.ModTimeNotSupported)
 }

+func TestWriteFileHandleCloseWithError(t *testing.T) {
+	// io.ErrUnexpectedEOF must fail the upload like any other reason.
+	for _, reason := range []error{errors.New("upload interrupted"), io.ErrUnexpectedEOF} {
+		t.Run(reason.Error(), func(t *testing.T) {
+			r, vfs, fh := writeHandleCreate(t)
+
+			// Write some data then abandon the write mid-stream
+			n, err := fh.Write([]byte("hello"))
+			require.NoError(t, err)
+			assert.Equal(t, 5, n)
+
+			err = fh.CloseWithError(reason)
+			require.Error(t, err)
+			assert.ErrorContains(t, err, reason.Error())
+
+			// Check double close
+			assert.Equal(t, ECLOSED, fh.Close())
+
+			if *fstest.RemoteName == "" {
+				_, err = vfs.Stat("file1")
+				assert.Equal(t, ENOENT, err, "The abandoned file must not exist in the VFS or on the remote")
+				fstest.CheckListingWithPrecision(t, r.Fremote, []fstest.Item{}, []string{}, fs.ModTimeNotSupported)
+			}
+		})
+	}
+
+	// CloseWithError with a nil reason behaves like Close and commits the file
+	t.Run("NilReason", func(t *testing.T) {
+		r, vfs := newTestVFS(t)
+		h, err := vfs.OpenFile("file2", os.O_WRONLY|os.O_CREATE, 0777)
+		require.NoError(t, err)
+		fh2, ok := h.(*WriteFileHandle)
+		require.True(t, ok)
+		_, err = fh2.Write([]byte("hello"))
+		require.NoError(t, err)
+		require.NoError(t, fh2.CloseWithError(nil))
+		file2 := fstest.NewItem("file2", "hello", t1)
+		fstest.CheckListingWithPrecision(t, r.Fremote, []fstest.Item{file2}, []string{}, fs.ModTimeNotSupported)
+	})
+}
+
 func TestWriteFileHandleFlush(t *testing.T) {
 	_, vfs, fh := writeHandleCreate(t)

@@ -383,3 +425,83 @@ func TestFileReadAtZeroLength(t *testing.T) {
 func TestFileReadAtNonZeroLength(t *testing.T) {
 	testFileReadAt(t, 100)
 }
+
+// stallingFs wraps an Fs so that streaming uploads accept a little data and
+// then stall until their context is cancelled, like a remote that has
+// stopped accepting data.
+type stallingFs struct {
+	fs.Fs
+	stalled  chan struct{} // closed once the upload has stalled
+	features *fs.Features
+}
+
+func newStallingFs(base fs.Fs) *stallingFs {
+	f := &stallingFs{Fs: base, stalled: make(chan struct{})}
+	features := *base.Features()
+	features.PutStream = f.PutStream
+	f.features = &features
+	return f
+}
+
+// Features returns the optional features of this Fs
+func (f *stallingFs) Features() *fs.Features { return f.features }
+
+// PutStream reads a little of in then stalls until ctx is cancelled.
+func (f *stallingFs) PutStream(ctx context.Context, in io.Reader, src fs.ObjectInfo, options ...fs.OpenOption) (fs.Object, error) {
+	buf := make([]byte, 1024)
+	_, _ = in.Read(buf)
+	close(f.stalled)
+	<-ctx.Done()
+	return nil, ctx.Err()
+}
+
+// TestWriteFileHandleCloseWithErrorStalled checks that CloseWithError
+// interrupts an upload the backend has stopped accepting data for. The
+// writer is blocked in the pipe holding the handle's lock, so the abandon
+// must fail the upload to unblock it, rather than waiting on the lock
+// until the backend gives up of its own accord.
+func TestWriteFileHandleCloseWithErrorStalled(t *testing.T) {
+	r := fstest.NewRun(t)
+	f := newStallingFs(r.Fremote)
+	// A tweaked option so this VFS is not deduplicated onto an active VFS
+	// of the unwrapped remote, whose config string it shares.
+	opt := vfscommon.Opt
+	opt.WriteWait += fs.Duration(time.Millisecond)
+	vfs := New(context.Background(), f, &opt)
+	t.Cleanup(vfs.Shutdown)
+
+	h, err := vfs.OpenFile("stalled", os.O_WRONLY|os.O_CREATE, 0777)
+	require.NoError(t, err)
+	fh, ok := h.(*WriteFileHandle)
+	require.True(t, ok)
+
+	// Write more than the transfer buffers absorb, so the write blocks on
+	// the stalled upload while holding the handle's lock.
+	writeDone := make(chan error, 1)
+	go func() {
+		_, err := fh.Write(make([]byte, 20<<20))
+		writeDone <- err
+	}()
+
+	select {
+	case <-f.stalled:
+	case <-time.After(10 * time.Second):
+		t.Fatal("upload never started")
+	}
+
+	// The abandon must interrupt the stalled upload promptly.
+	closeDone := make(chan error, 1)
+	go func() { closeDone <- fh.CloseWithError(errors.New("abandoned")) }()
+	select {
+	case err := <-closeDone:
+		require.Error(t, err)
+	case <-time.After(10 * time.Second):
+		t.Fatal("CloseWithError did not interrupt the stalled upload")
+	}
+	select {
+	case err := <-writeDone:
+		require.Error(t, err)
+	case <-time.After(10 * time.Second):
+		t.Fatal("the blocked write was never unblocked")
+	}
+}