ftp: fix ranged reads failing when the server answers 425 to an early close - Fixes #10200

A ranged read through the VFS opens the object at offset 0 and then
seeks, so the first RETR is closed before anything was read. Egnyte
answers that close with 425 instead of 426.

425 was not masked, so ftpReadCloser.Close quit the connection and
returned an error. The VFS then retried and closed the same reader
again; the second Close got nil from the already closed response, put
the connection that had been quit back in the pool and returned its
token a second time. Later operations failed with "use of closed
network connection" and serve s3 returned InternalError.

Now a second Close does nothing, a connection is never pooled after
Quit, and 425 is masked only when nothing was read, so a transfer cut
after data was read is still an error.

backend/ftp/ftp.go

@@ -1269,15 +1269,18 @@ func (o *Object) Storable() bool {
 
 // ftpReadCloser implements io.ReadCloser for FTP objects.
 type ftpReadCloser struct {
-	rc  io.ReadCloser
-	c   *ftp.ServerConn
-	f   *Fs
-	err error // errors found during read
+	rc     io.ReadCloser
+	c      *ftp.ServerConn
+	f      *Fs
+	err    error // errors found during read
+	closed bool  // set by Close so a second call is a no-op
+	read   int64 // bytes returned by Read
 }
 
 // Read bytes into p
 func (f *ftpReadCloser) Read(p []byte) (n int, err error) {
 	n, err = f.rc.Read(p)
+	f.read += int64(n)
 	if err != nil && err != io.EOF {
 		f.err = err // store any errors for Close to examine
 	}
@@ -1286,6 +1289,10 @@ func (f *ftpReadCloser) Read(p []byte) (n int, err error) {
 
 // Close the FTP reader and return the connection to the pool
 func (f *ftpReadCloser) Close() error {
+	if f.closed {
+		return nil
+	}
+	f.closed = true
 	var err error
 	errchan := make(chan error, 1)
 	go func() {
@@ -1309,6 +1316,7 @@ func (f *ftpReadCloser) Close() error {
 	// if errors while reading or closing, dump the connection
 	if err != nil || f.err != nil {
 		_ = f.c.Quit()
+		f.c = nil // never pool a connection after Quit
 		f.f.putFtpConnection(nil, nil)
 	} else {
 		f.f.putFtpConnection(&f.c, nil)
@@ -1320,6 +1328,12 @@ func (f *ftpReadCloser) Close() error {
 		switch errX.Code {
 		case ftp.StatusTransfertAborted, ftp.StatusFileUnavailable, ftp.StatusAboutToSend, ftp.StatusRequestedFileActionOK:
 			err = nil
+		case ftp.StatusCanNotOpenDataConnection:
+			// Egnyte answers 425 when a RETR is closed before the first
+			// read; a transfer cut after data was read stays an error
+			if f.read == 0 {
+				err = nil
+			}
 		}
 	}
 	return err

backend/ftp/ftp_internal_test.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io"
 	"net"
 	"net/textproto"
 	"strings"
@@ -244,3 +245,65 @@ func TestPutFtpConnectionDiscardsDeadConnection(t *testing.T) {
 		assert.Empty(t, f.pool)
 	})
 }
+
+// fakeDataReader stands in for the RETR data stream of ftpReadCloser.
+type fakeDataReader struct {
+	data     []byte
+	closeErr error
+}
+
+func (r *fakeDataReader) Read(p []byte) (int, error) {
+	if len(r.data) == 0 {
+		return 0, io.EOF
+	}
+	n := copy(p, r.data)
+	r.data = r.data[n:]
+	return n, nil
+}
+
+func (r *fakeDataReader) Close() error { return r.closeErr }
+
+// TestFtpReadCloserClose checks that an early-closed RETR answered with 425
+// is masked only when nothing was read, that a connection which got QUIT is
+// never pooled, and that a second Close does not touch the pool.
+func TestFtpReadCloserClose(t *testing.T) {
+	err425 := &textproto.Error{Code: ftp.StatusCanNotOpenDataConnection, Msg: "Can't open data connection."}
+	newReader := func(t *testing.T, data []byte, closeErr error) (*ftpReadCloser, *Fs) {
+		server := startFakeFTPServer(t)
+		c, err := ftp.Dial(server.ln.Addr().String())
+		require.NoError(t, err)
+		f := &Fs{opt: Options{}}
+		return &ftpReadCloser{rc: &fakeDataReader{data: data, closeErr: closeErr}, c: c, f: f}, f
+	}
+
+	t.Run("425 before any read is masked and the connection is not pooled", func(t *testing.T) {
+		rc, f := newReader(t, nil, err425)
+		assert.NoError(t, rc.Close())
+		assert.Empty(t, f.pool)
+		assert.NoError(t, rc.Close())
+		assert.Empty(t, f.pool)
+	})
+
+	t.Run("425 after data was read stays an error", func(t *testing.T) {
+		rc, f := newReader(t, []byte("partial"), err425)
+		buf := make([]byte, 4)
+		n, err := rc.Read(buf)
+		require.NoError(t, err)
+		require.Equal(t, 4, n)
+		err = rc.Close()
+		require.Error(t, err)
+		assert.Equal(t, ftp.StatusCanNotOpenDataConnection, textprotoError(err).Code)
+		assert.Empty(t, f.pool)
+	})
+
+	t.Run("clean close pools the connection once", func(t *testing.T) {
+		rc, f := newReader(t, []byte("data"), nil)
+		_, err := io.ReadAll(rc)
+		require.NoError(t, err)
+		assert.NoError(t, rc.Close())
+		require.Len(t, f.pool, 1)
+		assert.NoError(t, rc.Close())
+		assert.Len(t, f.pool, 1)
+		_ = f.pool[0].Quit()
+	})
+}