diff --git a/mitmproxy/addons/proxyserver.py b/mitmproxy/addons/proxyserver.py
index 3f5f1a445..549d05739 100644
--- a/mitmproxy/addons/proxyserver.py
+++ b/mitmproxy/addons/proxyserver.py
@@ -5,7 +5,7 @@ from typing import Dict, Optional, Tuple
 from mitmproxy import command, controller, ctx, flow, http, log, master, options, platform, tcp, websocket
 from mitmproxy.flow import Error, Flow
 from mitmproxy.proxy import commands, events
-from mitmproxy.proxy import server, server_hooks
+from mitmproxy.proxy import server
 from mitmproxy.proxy.layers.tcp import TcpMessageInjected
 from mitmproxy.proxy.layers.websocket import WebSocketMessageInjected
 from mitmproxy.utils import asyncio_utils, human, strutils
@@ -180,8 +180,3 @@ class Proxyserver:
             self.inject_event(event)
         except ValueError as e:
             ctx.log.warn(str(e))
-
-    def server_connected(self, ctx: server_hooks.ServerConnectionHookData):
-        # check if the outbound part of this connection appeared as a new client.
-        if ctx.server.sockname in self._connections:
-            ctx.server.error = "Stopped mitmproxy from recursively connecting to itself."
diff --git a/mitmproxy/proxy/server.py b/mitmproxy/proxy/server.py
index 4e3796129..a57071274 100644
--- a/mitmproxy/proxy/server.py
+++ b/mitmproxy/proxy/server.py
@@ -177,31 +177,32 @@ class ConnectionHandler(metaclass=abc.ABCMeta):
                 else:
                     addr = human.format_address(command.connection.address)
                 self.log(f"server connect {addr}")
-                await self.handle_hook(server_hooks.ServerConnectedHook(hook_data))
+                connected_hook = asyncio_utils.create_task(
+                    self.handle_hook(server_hooks.ServerConnectedHook(hook_data)),
+                    name=f"handle_hook(server_connected) {addr}",
+                    client=self.client.peername,
+                )
+                if not connected_hook:
+                    return  # this should not be needed, see asyncio_utils.create_task
 
-                if errmsg := command.connection.error:
-                    self.log(f"server connection to {addr} killed: {errmsg}")
-                    self.server_event(events.OpenConnectionCompleted(command, f"Connection killed: {errmsg}"))
-                    del self.transports[command.connection]
-                    writer.close()
-                else:
-                    self.server_event(events.OpenConnectionCompleted(command, None))
+                self.server_event(events.OpenConnectionCompleted(command, None))
 
-                    # during connection opening, this function is the designated handler that can be cancelled.
-                    # once we have a connection, we do want the teardown here to happen in any case, so we
-                    # reassign the handler to .handle_connection and then clean up here once that is done.
-                    new_handler = asyncio_utils.create_task(
-                        self.handle_connection(command.connection),
-                        name=f"server connection handler for {addr}",
-                        client=self.client.peername,
-                    )
-                    if not new_handler:
-                        return  # this should not be needed, see asyncio_utils.create_task
-                    self.transports[command.connection].handler = new_handler
-                    await asyncio.wait([new_handler])
+                # during connection opening, this function is the designated handler that can be cancelled.
+                # once we have a connection, we do want the teardown here to happen in any case, so we
+                # reassign the handler to .handle_connection and then clean up here once that is done.
+                new_handler = asyncio_utils.create_task(
+                    self.handle_connection(command.connection),
+                    name=f"server connection handler for {addr}",
+                    client=self.client.peername,
+                )
+                if not new_handler:
+                    return  # this should not be needed, see asyncio_utils.create_task
+                self.transports[command.connection].handler = new_handler
+                await asyncio.wait([new_handler])
 
                 self.log(f"server disconnect {addr}")
                 command.connection.timestamp_end = time.time()
+                await connected_hook  # wait here for this so that closed always comes after connected.
                 await self.handle_hook(server_hooks.ServerDisconnectedHook(hook_data))
 
     async def handle_connection(self, connection: Connection) -> None: