From ee0a70f875e8fa17239d7f04fc80dc64be4c019b Mon Sep 17 00:00:00 2001 From: Nick Craig-Wood Date: Tue, 6 Oct 2026 10:29:28 +0100 Subject: [PATCH] serve ftp: fix transfers failing after 5 minutes with --auth-proxy With --auth-proxy the backend of each user is shut down once it has been unused for 5 minutes. Only the start of each FTP command counted as a use, so an upload or download lasting longer than that had its backend shut down under it and failed with "context canceled". Each FTP command now holds the backend it uses until it has finished, which for a download is when the file is closed. This was introduced in v1.75.1 by f425f8d46 serve: refactor VFS and proxy handling into Provider --- cmd/serve/ftp/ftp.go | 54 ++++++++++++++++++--- cmd/serve/ftp/ftp_test.go | 99 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 147 insertions(+), 6 deletions(-) diff --git a/cmd/serve/ftp/ftp.go b/cmd/serve/ftp/ftp.go index 6783a51e1..b8902256f 100644 --- a/cmd/serve/ftp/ftp.go +++ b/cmd/serve/ftp/ftp.go @@ -16,6 +16,7 @@ import ( "regexp" "strconv" "strings" + "sync" "time" "github.com/rclone/rclone/cmd" @@ -358,13 +359,27 @@ func (d *driver) CheckPasswd(sctx *ftp.Context, user, pass string) (ok bool, err return true, nil } -// getVFS returns the VFS for this connection. +// getVFS returns the VFS for this connection, held so it can't be +// shut down while it is in use. The caller must call Shutdown on it +// when it has finished with it. // -// In proxy mode, getVFS calls proxy.Call on each FTP command which refreshes -// the proxy cache timer (like http/webdav). Therefore, connection-level pinning -// is not used; only individual transfers exceeding the cache expiry window -// could be affected. +// In proxy mode, getVFS calls proxy.Call on each FTP command which +// refreshes the proxy cache timer (like http/webdav). The proxy shuts +// the VFS down when that timer expires, which a single transfer can +// outlast, so the timer alone isn't enough to keep the VFS alive. func (d *driver) getVFS(sctx *ftp.Context) (VFS *vfs.VFS, err error) { + VFS, err = d.findVFS(sctx) + if err != nil { + return nil, err + } + if !VFS.Hold() { + return nil, errors.New("VFS has been shut down") + } + return VFS, nil +} + +// findVFS returns the VFS for this connection without holding it. +func (d *driver) findVFS(sctx *ftp.Context) (VFS *vfs.VFS, err error) { if !d.provider.IsProxy() { // If no proxy always use the same VFS return d.provider.VFS(), nil @@ -392,6 +407,7 @@ func (d *driver) Stat(sctx *ftp.Context, path string) (fi iofs.FileInfo, err err if err != nil { return nil, err } + defer VFS.Shutdown() n, err := VFS.Stat(path) if err != nil { return nil, err @@ -406,6 +422,7 @@ func (d *driver) ChangeDir(sctx *ftp.Context, path string) (err error) { if err != nil { return err } + defer VFS.Shutdown() n, err := VFS.Stat(path) if err != nil { return err @@ -423,6 +440,7 @@ func (d *driver) ListDir(sctx *ftp.Context, path string, callback func(iofs.File if err != nil { return err } + defer VFS.Shutdown() node, err := VFS.Stat(path) if err == vfs.ENOENT { return errors.New("directory not found") @@ -461,6 +479,7 @@ func (d *driver) DeleteDir(sctx *ftp.Context, path string) (err error) { if err != nil { return err } + defer VFS.Shutdown() node, err := VFS.Stat(path) if err != nil { return err @@ -482,6 +501,7 @@ func (d *driver) DeleteFile(sctx *ftp.Context, path string) (err error) { if err != nil { return err } + defer VFS.Shutdown() node, err := VFS.Stat(path) if err != nil { return err @@ -503,6 +523,7 @@ func (d *driver) Rename(sctx *ftp.Context, oldName, newName string) (err error) if err != nil { return err } + defer VFS.Shutdown() return VFS.Rename(oldName, newName) } @@ -513,6 +534,7 @@ func (d *driver) MakeDir(sctx *ftp.Context, path string) (err error) { if err != nil { return err } + defer VFS.Shutdown() dir, leaf, err := VFS.StatParent(path) if err != nil { return err @@ -528,6 +550,12 @@ func (d *driver) GetFile(sctx *ftp.Context, path string, offset int64) (size int if err != nil { return 0, nil, err } + // The returned file takes over the hold on the VFS + defer func() { + if err != nil { + VFS.Shutdown() + } + }() node, err := VFS.Stat(path) if err == vfs.ENOENT { fs.Infof(path, "File not found") @@ -552,7 +580,20 @@ func (d *driver) GetFile(sctx *ftp.Context, path string, offset int64) (size int tr := accounting.GlobalStats().NewTransferRemoteSize(path, node.Size(), d.f, nil) defer tr.Done(d.ctx, nil) - return node.Size(), handle, nil + return node.Size(), &heldFile{ReadCloser: handle, VFS: VFS}, nil +} + +// heldFile is an open file which holds its VFS until it is closed. +type heldFile struct { + io.ReadCloser + VFS *vfs.VFS + release sync.Once +} + +// Close closes the file and releases the hold on the VFS. +func (f *heldFile) Close() error { + defer f.release.Do(f.VFS.Shutdown) + return f.ReadCloser.Close() } // PutFile upload a file @@ -564,6 +605,7 @@ func (d *driver) PutFile(sctx *ftp.Context, path string, data io.Reader, offset if err != nil { return 0, err } + defer VFS.Shutdown() fi, err := VFS.Stat(path) if err == nil { isExist = true diff --git a/cmd/serve/ftp/ftp_test.go b/cmd/serve/ftp/ftp_test.go index 3e806cbda..119b0683d 100644 --- a/cmd/serve/ftp/ftp_test.go +++ b/cmd/serve/ftp/ftp_test.go @@ -9,8 +9,14 @@ package ftp import ( "context" + "io" + "os" + "path/filepath" + "sync" "testing" + "time" + ftpclient "github.com/jlaffaye/ftp" _ "github.com/rclone/rclone/backend/local" "github.com/rclone/rclone/cmd/serve/proxy" "github.com/rclone/rclone/cmd/serve/servetest" @@ -19,6 +25,7 @@ import ( "github.com/rclone/rclone/fs/config/obscure" "github.com/rclone/rclone/fs/rc" "github.com/rclone/rclone/lib/israce" + "github.com/rclone/rclone/lib/random" "github.com/rclone/rclone/vfs" "github.com/rclone/rclone/vfs/vfscommon" "github.com/stretchr/testify/assert" @@ -149,3 +156,95 @@ func TestNewServerError(t *testing.T) { assert.Nil(t, d) assert.Equal(t, before, vfs.ActiveCount(), "VFS leaked after failed server creation") } + +// TestAuthProxyTransferOutlivesCache checks transfers in progress +// carry on working when the auth proxy drops their VFS from its cache, +// as it does when a transfer takes longer than the cache expiry time. +func TestAuthProxyTransferOutlivesCache(t *testing.T) { + const addr = "127.0.0.1:" + testPORT + root := t.TempDir() + contents := random.String(32 * 1024 * 1024) + require.NoError(t, os.WriteFile(filepath.Join(root, "download.bin"), []byte(contents), 0666)) + + prog, err := filepath.Abs("../servetest/proxy_code.go") + require.NoError(t, err) + opt := Opt + opt.ListenAddr = addr + opt.PassivePorts = testPASSIVEPORTRANGE + proxyOpt := proxy.Opt + proxyOpt.AuthProxy = "go run " + prog + " " + root + d, err := newServer(context.Background(), nil, &opt, &vfscommon.Opt, &proxyOpt) + require.NoError(t, err) + quit := make(chan struct{}) + go func() { + assert.NoError(t, d.Serve()) + close(quit) + }() + defer func() { + assert.NoError(t, d.Shutdown()) + <-quit + }() + + var c *ftpclient.ServerConn + require.Eventually(t, func() bool { + c, err = ftpclient.Dial(addr) + return err == nil + }, 10*time.Second, 10*time.Millisecond) + defer func() { _ = c.Quit() }() + require.NoError(t, c.Login(testUSER, testPASS)) + + // Only the IP of the address is used, which is the same as the client's + _, vfsKey, err := d.provider.Proxy().Call(testUSER, testPASS, false, addr) + require.NoError(t, err) + + // expire waits for a transfer to be using the VFS, then drops + // everything from the proxy's cache as if it had expired. + expire := func(t *testing.T) { + var VFS *vfs.VFS + require.Eventually(t, func() bool { + VFS = d.provider.Proxy().Get(vfsKey) + return VFS != nil && VFS.Stats()["inUse"] == int32(2) + }, 10*time.Second, 10*time.Millisecond, "transfer isn't holding the VFS") + d.provider.Proxy().Shutdown() + assert.Equal(t, int32(1), VFS.Stats()["inUse"], "VFS not held by the transfer alone") + } + + t.Run("Download", func(t *testing.T) { + resp, err := c.Retr("download.bin") + require.NoError(t, err) + defer func() { _ = resp.Close() }() + start := make([]byte, 1024) + _, err = io.ReadFull(resp, start) + require.NoError(t, err) + expire(t) + rest, err := io.ReadAll(resp) + require.NoError(t, err) + require.NoError(t, resp.Close()) + assert.True(t, contents == string(start)+string(rest), "download corrupted") + }) + + t.Run("Upload", func(t *testing.T) { + pr, pw := io.Pipe() + var storErr error + var wg sync.WaitGroup + wg.Go(func() { + storErr = c.Stor("upload.bin", pr) + }) + // Finish the upload before the connection is used again + defer func() { + _ = pw.Close() + wg.Wait() + }() + _, err := io.WriteString(pw, contents[:1024*1024]) + require.NoError(t, err) + expire(t) + _, err = io.WriteString(pw, contents[1024*1024:]) + require.NoError(t, err) + require.NoError(t, pw.Close()) + wg.Wait() + require.NoError(t, storErr) + got, err := os.ReadFile(filepath.Join(root, "upload.bin")) + require.NoError(t, err) + assert.True(t, contents == string(got), "upload corrupted") + }) +}