Make data transfer timeout tracking independent of streams controlled by

the user stalling.

Could address #256

Author: patrickjuchli

README.md

@@ -67,7 +67,11 @@ client.ftp.verbose = true
 
 `new Client(timeout, options)`
 
-Create a client instance. Configure it with a timeout in milliseconds that will be used for any connection made. Use 0 to disable timeouts, default is 30 seconds. Options are: 
+Create a client instance. Configure it with a timeout in milliseconds that will be used for any connection made. Use 0 to disable timeouts, default is 30 seconds.
+
+The timeout applies to the server: a transfer fails if the server stops making progress for that long. It doesn't limit how long your own streams may take. A download piped into a slow destination, or an upload fed by a slow source, can hold up a transfer for as long as it needs to without running into a timeout. If you want to limit that as well, do it in your own code.
+
+Options are: 
 
 - `allowSeparateTransferHost (boolean)`, the FTP spec makes it possible for a server to tell the client to use a different IP address for file transfers than for the initial control connection. This is a potential vector for FTP bounce attacks, so by default this is set to `false` and the library will throw an error if a server tries to redirect transfers to a different host. Set this to `true` only if you are connecting to a server that legitimately requires it.
 

src/TransferWatchdog.ts

@@ -0,0 +1,99 @@
+import { Socket } from "net"
+
+/** A transfer either sends data to or receives data from the data connection. */
+export type TransferDirection = "upload" | "download"
+
+/** How long to wait between two checks of a transfer at most, in milliseconds. */
+const maxCheckIntervalMs = 500
+
+/**
+ * How many checks a timeout is split into, at least. A transfer is only reported as stalled if
+ * every single one of them found the connection idle. Deciding on a single check would make brief
+ * moments where a transfer looks idle without being stalled matter, for example just after a
+ * destination signalled that it can accept data again but before the first bytes arrived.
+ *
+ * Splitting into n checks means such a moment has to last about (n-1)/n of the timeout to be
+ * mistaken for a stall, so the value matters much less than it being greater than one. It only
+ * has an effect on timeouts below `minChecksPerTimeout * maxCheckIntervalMs` anyway, longer ones
+ * are split into more checks by the interval limit alone.
+ */
+const minChecksPerTimeout = 4
+
+/**
+ * Watches a data connection during a transfer and reports it as stalled if the server stopped
+ * making progress.
+ *
+ * This replaces a plain inactivity timeout on the data socket. Such a timeout can't tell apart
+ * "the server stopped sending" from "our own source or destination isn't ready yet". A slow local
+ * stream is not an error: A download piped into a decompressor, or an upload fed by a stream that
+ * computes its data, can legitimately leave the connection idle for minutes. Timing out on that
+ * kills a healthy transfer and truncates the data the destination already received.
+ *
+ * Only time spent waiting for the server counts towards the timeout, see `isWaitingForServer`.
+ */
+export class TransferWatchdog {
+
+    protected timer: NodeJS.Timeout | undefined = undefined
+
+    /**
+     * Start watching a transfer. Calls `onStall` if the server hasn't made progress for
+     * `timeout` milliseconds. A timeout of 0 disables the watchdog.
+     */
+    start(socket: Socket, direction: TransferDirection, timeout: number, onStall: () => void) {
+        this.stop()
+        if (timeout <= 0) {
+            return
+        }
+        const intervalMs = Math.max(1, Math.min(Math.floor(timeout / minChecksPerTimeout), maxCheckIntervalMs))
+        let lastBytes = countBytes(socket)
+        let idleMs = 0
+        this.timer = setInterval(() => {
+            const bytes = countBytes(socket)
+            const madeProgress = bytes !== lastBytes
+            lastBytes = bytes
+            if (madeProgress || !isWaitingForServer(socket, direction)) {
+                idleMs = 0
+                return
+            }
+            // Count checks instead of measuring elapsed time: a blocked event loop delays our
+            // checks just as much as it delays reading from the socket, and that's not something
+            // the server should be blamed for.
+            idleMs += intervalMs
+            if (idleMs >= timeout) {
+                this.stop()
+                onStall()
+            }
+        }, intervalMs)
+        // Don't keep the process alive just to watch a transfer.
+        this.timer.unref()
+    }
+
+    /**
+     * Stop watching. Safe to call at any time, also if no transfer is being watched.
+     */
+    stop() {
+        if (this.timer) {
+            clearInterval(this.timer)
+            this.timer = undefined
+        }
+    }
+}
+
+function countBytes(socket: Socket): number {
+    return socket.bytesRead + socket.bytesWritten
+}
+
+/**
+ * Returns true if the transfer can only continue once the server acts. When downloading, that's
+ * the case as long as we're ready to receive: if the socket is paused, our destination applied
+ * backpressure and the server may well be waiting for us. When uploading, it's the case if we
+ * still have data queued that the server isn't accepting: an empty queue means we're waiting for
+ * our own source instead.
+ *
+ * `isPaused()` only reports backpressure for a socket that is being piped, which is what
+ * `downloadTo` in transfer.ts does. Should that ever change to reading the socket directly, e.g.
+ * by iterating over it, this would report a transfer as stalled while it's waiting for us.
+ */
+function isWaitingForServer(socket: Socket, direction: TransferDirection): boolean {
+    return direction === "download" ? !socket.isPaused() : socket.writableLength > 0
+}

src/transfer.ts

@@ -1,8 +1,10 @@
 import { EventEmitter } from "events"
+import { Socket } from "net"
 import { Readable, Writable, pipeline } from "stream"
 import { TLSSocket, connect as connectTLS } from "tls"
 import { FTPContext, FTPResponse, TaskResolver } from "./FtpContext"
 import { ProgressTracker, ProgressType } from "./ProgressTracker"
+import { TransferWatchdog } from "./TransferWatchdog"
 import { describeAddress, describeTLS, ipIsPrivateV4Address, isLoopback } from "./netUtils"
 import { positiveCompletion, positiveIntermediate } from "./parseControlResponse"
 
@@ -109,7 +111,8 @@ export function parsePasvResponse(message: string): { host: string, port: number
 
 export function connectForPassiveTransfer(host: string, port: number, ftp: FTPContext): Promise<void> {
     return new Promise((resolve, reject) => {
-        let socket = ftp._newSocket()
+        const rawSocket = ftp._newSocket()
+        let socket: Socket | TLSSocket = rawSocket
         const handleConnErr = function(err: Error) {
             err.message = "Can't open data connection in passive mode: " + err.message
             reject(err)
@@ -123,8 +126,8 @@ export function connectForPassiveTransfer(host: string, port: number, ftp: FTPCo
         socket.on("timeout", handleTimeout)
         socket.connect({ port, host, family: ftp.ipFamily}, () => {
             if (ftp.socket instanceof TLSSocket) {
-                socket = connectTLS(Object.assign({}, ftp.tlsOptions, {
-                    socket,
+                const tlsSocket = connectTLS(Object.assign({}, ftp.tlsOptions, {
+                    socket: rawSocket,
                     // Reuse the TLS session negotiated earlier when the control connection
                     // was upgraded. Servers expect this because it provides additional
                     // security: If a completely new session would be negotiated, a hacker
@@ -135,7 +138,8 @@ export function connectForPassiveTransfer(host: string, port: number, ftp: FTPCo
                 // When the server issues a new session ticket after this data connection's
                 // TLS handshake (TLS 1.3 single-use tickets), capture it so the next data
                 // connection can present a fresh ticket and resume successfully.
-                socket.on("session", session => { ftp.tlsSessionStore = session })
+                tlsSocket.on("session", session => { ftp.tlsSessionStore = session })
+                socket = tlsSocket
                 // It's the responsibility of the transfer task to wait until the
                 // TLS socket issued the event 'secureConnect'. We can't do this
                 // here because some servers will start upgrading after the
@@ -144,6 +148,11 @@ export function connectForPassiveTransfer(host: string, port: number, ftp: FTPCo
                 // is ready. But for upload this has to be taken into account,
                 // see the details in the upload() function below.
             }
+            // Disable the timeout that was guarding the connection attempt. This has to happen on
+            // the socket it was set on: when using TLS, `socket` is by now a wrapper around that
+            // socket, and a timeout left running underneath would destroy the data connection
+            // during a transfer that is idle for a legitimate reason.
+            rawSocket.setTimeout(0)
             // Let the FTPContext listen to errors from now on, remove local handler.
             socket.removeListener("error", handleConnErr)
             socket.removeListener("timeout", handleTimeout)
@@ -165,6 +174,7 @@ class TransferResolver {
 
     protected response: FTPResponse | undefined = undefined
     protected dataTransferDone = false
+    protected readonly watchdog = new TransferWatchdog()
 
     /**
      * Instantiate a TransferResolver
@@ -178,31 +188,34 @@ class TransferResolver {
      * @param type - Type of transfer, usually "upload" or "download".
      */
     onDataStart(name: string, type: ProgressType) {
-        // Let the data socket be in charge of tracking timeouts during transfer.
+        // Let the data connection be in charge of tracking timeouts during transfer.
         // The control socket sits idle during this time anyway and might provoke
         // a timeout unnecessarily. The control connection will take care
         // of timeouts again once data transfer is complete or failed.
         if (this.ftp.dataSocket === undefined) {
             throw new Error("Data transfer should start but there is no data connection.")
         }
         this.ftp.socket.setTimeout(0)
-        this.ftp.dataSocket.setTimeout(this.ftp.timeout)
+        // An inactivity timeout on the data socket would also fire while a slow local source or
+        // destination is holding up an otherwise healthy transfer. Watch the transfer instead.
+        this.ftp.dataSocket.setTimeout(0)
+        this.watchdog.start(this.ftp.dataSocket, type === "upload" ? "upload" : "download", this.ftp.timeout, () => {
+            this.ftp.closeWithError(new Error("Timeout (data socket)"))
+        })
         this.progress.start(this.ftp.dataSocket, name, type)
     }
 
     /**
      * The data connection has finished the transfer.
      */
     onDataDone(task: TaskResolver) {
+        this.watchdog.stop()
         this.progress.updateAndStop()
         // Hand-over timeout tracking back to the control connection. It's possible that
         // we don't receive the response over the control connection that the transfer is
         // done. In this case, we want to correctly associate the resulting timeout with
         // the control connection.
         this.ftp.socket.setTimeout(this.ftp.timeout)
-        if (this.ftp.dataSocket) {
-            this.ftp.dataSocket.setTimeout(0)
-        }
         this.dataTransferDone = true
         this.tryResolve(task)
     }
@@ -219,6 +232,7 @@ class TransferResolver {
      * An error has been reported and the task should be rejected.
      */
     onError(task: TaskResolver, err: Error) {
+        this.watchdog.stop()
         this.progress.updateAndStop()
         this.ftp.socket.setTimeout(this.ftp.timeout)
         this.ftp.dataSocket = undefined
@@ -324,6 +338,10 @@ export function downloadTo(destination: Writable, config: TransferConfig): Promi
             }
             config.ftp.log(`Downloading from ${describeAddress(dataSocket)} (${describeTLS(dataSocket)})`)
             resolver.onDataStart(config.remotePath, config.type)
+            // Keep piping the data connection: TransferWatchdog recognizes a destination that
+            // can't keep up by the socket being paused, and only piping does that. Consuming the
+            // socket in another way, e.g. by iterating over it, makes the watchdog report a
+            // transfer as stalled while it's in fact waiting for us.
             pipeline(dataSocket, destination, err => {
                 if (err) {
                     resolver.onError(task, err)

test/MockFtpServer.js

@@ -8,6 +8,15 @@ const DEFAULT_HANDLERS = {
     quit: () => "200 Bye"
 }
 
+/**
+ * A client that goes away mid-conversation is a normal thing for a server to see, and several
+ * tests provoke exactly that. Without this, writing to such a connection throws EPIPE/ECONNRESET
+ * as an uncaught exception and fails whichever test happens to be running.
+ */
+function ignoreErrors(conn) {
+    conn.on("error", () => {})
+}
+
 module.exports = class MockFtpServer {
     constructor() {
         this.didOpenDataConn = () => {}
@@ -22,6 +31,7 @@ module.exports = class MockFtpServer {
             this.ctrlConn = conn
             this.connections.push(conn)
             conn.allowHalfOpen = true
+            ignoreErrors(conn)
             conn.write(`200 Welcome${NEWLINE}`)
             conn.on("data", data => {
                 const command = data.toString().trim()
@@ -41,6 +51,7 @@ module.exports = class MockFtpServer {
         this.dataServer = net.createServer(conn => {
             this.dataConn = conn
             this.connections.push(conn)
+            ignoreErrors(conn)
             this.didOpenDataConn()
             const bufs = []
             conn.on("data", data => {
@@ -58,6 +69,9 @@ module.exports = class MockFtpServer {
     }
 
     writeCtrl(payload) {
+        if (!this.ctrlConn || this.ctrlConn.destroyed) {
+            return
+        }
         this.ctrlConn.write(`${payload}${NEWLINE}`)
     }
 

test/downloadSpec.js

@@ -8,6 +8,8 @@ const fs = require("fs");
 
 const FILENAME = "file.txt"
 const TIMEOUT = 1000
+// Used where a test has to wait for a timeout to happen, or to not happen.
+const SHORT_TIMEOUT = 100
 const EMPTY_TEXT = ""
 const SHORT_TEXT = "Short"
 const MEDIUM_TEXT = "s".repeat(45017) // https://github.com/patrickjuchli/basic-ftp/issues/205
@@ -79,6 +81,50 @@ describe("Download to stream", () => {
         assert.deepEqual(buf.getText("utf-8"), MEDIUM_TEXT)
     })
 
+    // A destination that is slow to accept data is not a broken connection. Timing out on it
+    // aborts a healthy transfer and truncates whatever the destination received so far.
+    it("doesn't time out while a slow destination is holding up the transfer", async () => {
+        payload = "s".repeat(1000 * 1000)
+        client.ftp.timeout = SHORT_TIMEOUT
+        let received = 0
+        let blocking = false
+        let bytesReadWhileBlocked = 0
+        const destination = new Writable({
+            highWaterMark: 1,
+            write(chunk, enc, cb) {
+                received += chunk.length
+                if (blocking) {
+                    cb()
+                    return
+                }
+                blocking = true
+                setTimeout(() => {
+                    bytesReadWhileBlocked = client.ftp.dataSocket.bytesRead
+                    cb()
+                }, 5 * SHORT_TIMEOUT)
+            }
+        })
+        await client.downloadTo(destination, FILENAME)
+        assert.strictEqual(received, payload.length, "received all data")
+        assert.ok(bytesReadWhileBlocked < payload.length,
+            `transfer was really held up, read ${bytesReadWhileBlocked} of ${payload.length} bytes while blocked`)
+    })
+
+    it("times out if the server stops sending", async () => {
+        client.ftp.timeout = SHORT_TIMEOUT
+        server.addHandlers({
+            "pasv": () => `227 Entering Passive Mode (${server.dataAddressForPasvResponse})`,
+            // Send something, then go silent without ever closing the data connection.
+            "retr": () => {
+                setTimeout(() => server.dataConn.write("the beginning..."))
+                return "150 Ready to download"
+            }
+        })
+        return assert.rejects(() => client.downloadTo(new StringWriter(), FILENAME), {
+            message: "Timeout (data socket)"
+        })
+    })
+
     it("handles late destination stream error", async () => {
         server.addHandlers({
             "pasv": () => `227 Entering Passive Mode (${server.dataAddressForPasvResponse})`,
@@ -161,7 +207,32 @@ describe("Download to stream", () => {
         dataSocket.destroy(new Error("Error that should be ignored because task has completed successfully"))
     })
 
-    it.todo("stops tracking timeout after failure")
+    it("stops tracking timeout after failure", async () => {
+        client.ftp.timeout = SHORT_TIMEOUT
+        server.addHandlers({
+            "pasv": () => `227 Entering Passive Mode (${server.dataAddressForPasvResponse})`,
+            // Fail while the data connection is transferring.
+            "retr": () => {
+                setTimeout(() => {
+                    server.dataConn.write("the beginning...")
+                    server.writeCtrl("500 Something went wrong")
+                })
+                return "150 Ready to download"
+            },
+            "noop": () => "200 OK"
+        })
+        await assert.rejects(() => client.downloadTo(new StringWriter(), FILENAME), {
+            message: "500 Something went wrong"
+        })
+        assert.strictEqual(client.ftp.socket.timeout, 0, "control socket stopped tracking")
+        // Nothing may be left watching the failed transfer: it would report a timeout later on,
+        // taking down whatever the client is doing by then.
+        await client.access({ port: server.ctrlAddress.port, user: "test", password: "test" })
+        await new Promise(resolve => setTimeout(resolve, 3 * SHORT_TIMEOUT))
+        assert.strictEqual(client.closed, false, "client still connected after being idle")
+        await client.send("NOOP")
+    })
+
     it.todo("can get a directory listing")
     it.todo("uses control host IP if suggested data connection IP using PASV is private")
     it.todo("can download using TLS")