Handle connection closure gracefully and add MIT license

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2026 Daniel van Flymen
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

pyproject.toml

@@ -3,6 +3,7 @@ name = "synapse-p2p"
 dynamic = ["version"]
 description = "A language-agnostic network substrate for agent-to-agent collaboration"
 readme = "README.md"
+license = "MIT"
 keywords = ["agents", "agent-to-agent", "multi-agent", "rpc", "p2p", "msgpack", "substrate"]
 authors = [{ name = "Daniel van Flymen", email = "vanflymen@gmail.com" }]
 requires-python = ">=3.10"

synapse_p2p/node.py

@@ -186,6 +186,16 @@ async def handle_data(self, reader: asyncio.StreamReader, writer: asyncio.Stream
 
             result = await self._dispatch(rpc, connection)
             response = RPCResponse(id=request_id, ok=True, result=result)
+        except (asyncio.IncompleteReadError, ConnectionError, OSError):
+            logger.debug(
+                "Connection closed before a full request from {}:{}",
+                connection.ip,
+                connection.port,
+            )
+            writer.close()
+            with contextlib.suppress(ConnectionError, OSError):
+                await writer.wait_closed()
+            return
         except InvalidMessageError as e:
             logger.debug("Invalid message from {}:{}: {}", connection.ip, connection.port, e)
             response = RPCResponse(
@@ -201,10 +211,19 @@ async def handle_data(self, reader: asyncio.StreamReader, writer: asyncio.Stream
                 error=RPCError(code="internal_error", message=str(e)),
             )
 
-        await write_frame(writer, self.serializer_class.serialize(response))
-        logger.debug("Closing connection to {}:{}", connection.ip, connection.port)
-        writer.close()
-        await writer.wait_closed()
+        try:
+            await write_frame(writer, self.serializer_class.serialize(response))
+        except (ConnectionError, OSError):
+            logger.debug(
+                "Connection closed before response to {}:{}",
+                connection.ip,
+                connection.port,
+            )
+        finally:
+            logger.debug("Closing connection to {}:{}", connection.ip, connection.port)
+            writer.close()
+            with contextlib.suppress(ConnectionError, OSError):
+                await writer.wait_closed()
 
     def _print_startup(self) -> None:
         print(f"Listening on {self.bind}:{self.port} (advertising {self.address})\n")

synapse_p2p/tests/test_node.py

@@ -54,6 +54,29 @@ async def _send_rpc(node: Node, rpc: RemoteProcedureCall) -> RPCResponse:
     return await _send_message(node, rpc)
 
 
+@pytest.mark.asyncio
+async def test_node_ignores_connections_that_close_before_request():
+    node = Node(bind="127.0.0.1")
+    loop = asyncio.get_running_loop()
+    errors: list[dict] = []
+    previous_handler = loop.get_exception_handler()
+    loop.set_exception_handler(lambda _loop, context: errors.append(context))
+
+    tcp = await asyncio.start_server(node.handle_data, node.bind, node.port)
+    host, port = tcp.sockets[0].getsockname()[:2]
+
+    try:
+        async with tcp:
+            _reader, writer = await asyncio.open_connection(host, port)
+            writer.close()
+            await writer.wait_closed()
+            await asyncio.sleep(0)
+    finally:
+        loop.set_exception_handler(previous_handler)
+
+    assert errors == []
+
+
 @pytest.mark.asyncio
 async def test_node_round_trip_over_tcp():
     node = Node(bind="127.0.0.1")