diff --git a/src/clients/FtpClient.cc b/src/clients/FtpClient.cc
index 15a20b4b10c..7a48812a003 100644
--- a/src/clients/FtpClient.cc
+++ b/src/clients/FtpClient.cc
@@ -179,6 +179,14 @@ Ftp::DataChannel::addr(const Ip::Address &import)
     port = import.port();
 }
 
+void
+Ftp::DataChannel::appended(const size_t size)
+{
+    readBuf->appended(size);
+    payloadSeen += size;
+    debugs(9, 5, size << " bytes to readBuf, payloadSeen: " << payloadSeen);
+}
+
 /* Ftp::Client */
 
 Ftp::Client::Client(FwdState *fwdState):
@@ -979,8 +987,7 @@ Ftp::Client::dataRead(const CommIoCbParams &io)
     }
 
     if (io.flag == Comm::OK && io.size > 0) {
-        debugs(9, 5, "appended " << io.size << " bytes to readBuf");
-        data.readBuf->appended(io.size);
+        data.appended(io.size);
 #if USE_DELAY_POOLS
         DelayId delayId = entry->mem_obj->mostBytesAllowed();
         delayId.bytesIn(io.size);
diff --git a/src/clients/FtpClient.h b/src/clients/FtpClient.h
index a42e0e199a6..4428a78991a 100644
--- a/src/clients/FtpClient.h
+++ b/src/clients/FtpClient.h
@@ -99,11 +99,17 @@ class DataChannel: public Ftp::Channel
 
     void addr(const Ip::Address &addr); ///< import host and port
 
+    /// updates counters after this number of bytes have been added to readBuf
+    void appended(size_t size);
+
 public:
     MemBuf *readBuf;
     char *host;
     unsigned short port;
     bool read_pending;
+
+    /// amount of message payload/body received so far
+    int64_t payloadSeen = 0;
 };
 
 /// FTP client functionality shared among FTP Gateway and Relay clients.
diff --git a/src/clients/FtpGateway.cc b/src/clients/FtpGateway.cc
index 675c08dfec2..3c7ae0ac3b7 100644
--- a/src/clients/FtpGateway.cc
+++ b/src/clients/FtpGateway.cc
@@ -1008,6 +1008,11 @@ Ftp::Gateway::processReplyBody()
 
     entry->flush();
 
+    if (theSize >= 0 && data.payloadSeen >= theSize) {
+        markParsedVirginReplyAsWhole("whole virgin body");
+        completeForwarding();
+    }
+
     maybeReadVirginBody();
 }
 
@@ -2274,6 +2279,7 @@ ftpWriteTransferDone(Ftp::Gateway * ftpState)
     ftpState->entry->timestampsSet();   /* XXX Is this needed? */
     ftpState->markParsedVirginReplyAsWhole("ftpWriteTransferDone code 226 or 250");
     ftpSendReply(ftpState);
+    ftpState->completeForwarding();
 }
 
 static void
@@ -2639,7 +2645,7 @@ Ftp::Gateway::completeForwarding()
 {
     if (fwd == nullptr || flags.completed_forwarding) {
         debugs(9, 3, "avoid double-complete on FD " <<
-               (ctrl.conn ? ctrl.conn->fd : -1) << ", Data FD " << data.conn->fd <<
+               (ctrl.conn ? ctrl.conn->fd : -1) << ", Data FD " << (data.conn ? data.conn->fd : -1) <<
                ", this " << this << ", fwd " << fwd);
         return;
     }