ssh/Connection: postpone Destroy() call until DISCONNECT is flushed

If the connection is already encrypted, we need to wait for the worker
thread to finish encrypting the DISCONNECT packet, and only after that
can we send it to the socket.

This finishes the second part of commit 62f4ca0; it needs some
complicated code to ensure that the connection does no other I/O while
it's waiting for the flush.

src/Connection.cxx

@@ -988,6 +988,14 @@ void
 Connection::OnDisconnecting(SSH::DisconnectReasonCode reason_code,
 			    std::string_view msg) noexcept
 {
+	CConnection::OnDisconnecting(reason_code, msg);
+
+	/* some manual shutdown just in case the Destroy() is
+           postponed */
+	auth_timeout.Cancel();
+	socket_forward_listeners.clear_and_dispose(DeleteDisposer{});
+	occupied_task = {};
+
 	if (log_disconnect) {
 		log_disconnect = false;
 		LogFmt("Disconnecting: {}", msg);

src/OutgoingConnection.cxx

@@ -162,6 +162,8 @@ void
 OutgoingConnection::OnDisconnecting(SSH::DisconnectReasonCode reason_code,
 				    std::string_view msg) noexcept
 {
+	SSH::Connection::OnDisconnecting(reason_code, msg);
+
 	handler.OnOutgoingDisconnecting(reason_code, msg);
 }
 

src/ssh/CConnection.cxx

@@ -526,5 +526,18 @@ CConnection::OnWriteUnblocked() noexcept
 		if (i != nullptr)
 			i->OnWriteUnblocked();
 }
+void
+CConnection::OnDisconnecting(DisconnectReasonCode reason_code,
+			     std::string_view msg) noexcept
+{
+	GConnection::OnDisconnecting(reason_code, msg);
+
+	/* delete all channels so they don't try to do any I/O while
+           we're waiting for the DISCONNECT to be flushed */
+	for (auto &i : channels) {
+		delete i;
+		i = nullptr;
+	}
+}
 
 } // namespace SSH

src/ssh/CConnection.hxx

@@ -144,6 +144,8 @@ protected:
 			  std::span<const std::byte> payload) override;
 	void OnWriteBlocked() noexcept override;
 	void OnWriteUnblocked() noexcept override;
+	void OnDisconnecting(DisconnectReasonCode reason_code,
+			     std::string_view msg) noexcept;
 };
 
 } // namespace SSH

src/ssh/Connection.cxx

@@ -106,10 +106,29 @@ Connection::SendPacket(MessageNumber msg, std::span<const std::byte> payload)
 void
 Connection::DoDisconnect(DisconnectReasonCode reason_code, std::string_view msg) noexcept
 {
+	if (IsDead())
+		return;
+
 	OnDisconnecting(reason_code, msg);
 
 	SendPacket(MakeDisconnect(reason_code, msg));
 
+	if (output.IsEncrypted()) {
+		/* we have to wait for the worker thread to encrypt
+                   the the DISCONNECT packet before we can actually
+                   send it to the socket; therefore postpone the
+                   Destroy() call */
+		dead = true;
+
+		/* we now have very little patience with this
+                   client */
+		socket.SetWriteTimeout(std::chrono::seconds{1});
+
+		/* we don't want to receive anything from it */
+		socket.UnscheduleRead();
+		return;
+	}
+
 	try {
 		/* attempt to flush the DISCONNECT packet immediately
                    before we close the socket */
@@ -598,6 +617,11 @@ Connection::OnBufferedWrite()
 
 	switch (output.Flush()) {
 	case Output::FlushResult::DONE:
+		if (IsDead() && output.IsEmpty()) {
+			Destroy();
+			return false;
+		}
+
 		socket.UnscheduleWrite();
 		break;
 
@@ -629,6 +653,9 @@ Connection::OnBufferedError([[maybe_unused]] std::exception_ptr e) noexcept
 bool
 Connection::OnInputReady() noexcept
 try {
+	if (IsDead())
+		return false;
+
 	while (true) {
 		const auto payload = input.ReadPacket();
 		if (payload.data() == nullptr)

src/ssh/Connection.hxx

@@ -53,6 +53,12 @@ class Connection : BufferedSocketHandler, InputHandler
 
 	const Role role;
 
+	/**
+	 * If true, then the connection is about to be closed, only
+         * waiting for the DISCONNECT to be encrypted and sent.
+	 */
+	bool dead = false;
+
 	bool version_exchanged = false;
 
 	bool authenticated = false;
@@ -103,6 +109,10 @@ public:
 		metrics = &_metrics;
 	}
 
+	bool IsDead() const noexcept {
+		return dead;
+	}
+
 	[[gnu::pure]]
 	bool IsEncrypted() const noexcept;
 

src/ssh/GConnection.cxx

@@ -166,4 +166,15 @@ GConnection::HandlePacket(MessageNumber msg,
 	}
 }
 
+void
+GConnection::OnDisconnecting(DisconnectReasonCode reason_code,
+			     std::string_view msg) noexcept
+{
+	Connection::OnDisconnecting(reason_code, msg);
+
+	/* cancel all pending requests so they don't try to do any I/O
+           while we're waiting for the DISCONNECT to be flushed */
+	pending_global_requests.clear_and_dispose(DeleteDisposer{});
+}
+
 } // namespace SSH

src/ssh/GConnection.hxx

@@ -55,6 +55,8 @@ protected:
 	/* virtual methods from class SSH::Connection */
 	void HandlePacket(MessageNumber msg,
 			  std::span<const std::byte> payload) override;
+	void OnDisconnecting(DisconnectReasonCode reason_code,
+			     std::string_view msg) noexcept;
 };
 
 } // namespace SSH

src/ssh/Output.hxx

@@ -95,6 +95,13 @@ public:
 		return push_cipher != nullptr;
 	}
 
+	[[gnu::pure]]
+	bool IsEmpty() noexcept {
+		const std::scoped_lock lock{mutex};
+		return pending_queue.empty() && plain_queue.empty() &&
+		       next_plain_queue.empty() && encrypted_queue.empty();
+	}
+
 	const Cipher *GetCipher() const noexcept {
 		return push_cipher;
 	}