@@ -10,3 +10,70 @@ You can consider the following:
 - Use shorter non-asynchronous processing loops.
 - Use ``asyncio.sleep()`` instead of ``time.sleep()``.
 - Use a stable network connection.
+
+Connection recovery
+-------------------
+
+Hydrogram coordinates recovery through one task per MTProto session. A closed
+transport, connection reset, failed write, request timeout, or missing ping reply
+starts recovery. Concurrent requests wait for that same task. Recovery preserves
+the authorization key and never logs out, revokes a session, or deletes a session
+file.
+
+Connection attempts use exponential backoff with jitter. The session allows five
+consecutive initialization attempts, including connections that fail again within
+30 seconds. Backoff starts between 0.5 and 1 second, doubles between attempts, and
+is capped at 30 seconds. Each initialization can make up to three transport
+connection attempts. A connection that remains healthy for 30 seconds restores the
+recovery budget. Exhaustion leaves the session in the ``failed`` state and raises
+``ConnectionError`` to callers. Further RPCs fail promptly instead of silently
+starting another unlimited retry loop; an explicit session start or client
+reconnection is needed to try again.
+
+A lack of updates is normal for an idle client. Periodic pings require a pong and
+provide the transport liveness check. The update watchdog can probe update state
+after an idle period, but inactivity alone does not trigger a reconnect.
+
+Request outcomes
+----------------
+
+Requests that have not been sent can proceed after recovery. Requests already
+sent either return their available response or fail promptly with
+:class:`~hydrogram.errors.ConnectionLostError`, a ``ConnectionError`` subclass.
+A request whose response times out raises
+:class:`~hydrogram.errors.RequestTimeoutError`, a ``TimeoutError`` subclass.
+Both expose ``delivery_uncertain``: when true, Telegram may have processed the
+request. Check its outcome before deciding whether to retry a write operation.
+Hydrogram does not automatically replay these requests or RPCs returning server
+errors, since repeating an arbitrary method could duplicate side effects.
+
+An explicit bad-server-salt rejection permits a bounded resend with the corrected
+salt. Sequence-number errors rebuild only the ephemeral MTProto session and
+surface the rejected request to its caller. Authentication failures such as
+``AUTH_KEY_DUPLICATED`` or ``SESSION_REVOKED`` stop recovery and surface the error
+without changing stored credentials. Clock-related bad-message errors require
+correcting the local clock; they are not retried automatically. Two-factor login
+prompts remain available on the current connection.
+
+Connection diagnostics
+----------------------
+
+``app.connection_health`` returns a snapshot without making a network request:
+
+.. code-block:: python
+
+    health = app.connection_health
+    print(health["state"], health["is_connected"], health["last_error"])
+
+The snapshot contains ``state`` (``stopped``, ``connecting``, ``reconnecting``,
+``connected`` or ``failed``), ``is_connected``, ``dc_id``, ``reconnect_attempt``,
+``pending_requests``, ``last_received_at`` and ``last_error``. The last-received
+timestamp includes protocol traffic such as pongs, not just updates.
+``app.is_connected`` continues to describe the client lifecycle; use the health
+snapshot to inspect transport readiness during recovery.
+
+The ``hydrogram.session.session`` logger records connection loss, retry delays,
+successful recovery, and exhausted attempts. Retry records include ``dc_id``,
+``connection_state``, ``reconnect_attempt`` and ``reconnect_delay`` as structured
+logging fields. Stopping the client cancels its recovery, ping, receive, update
+watchdog, and owned update-processing tasks.
@@ -48,6 +48,7 @@
     BadRequest,
     CDNFileHashMismatch,
     ChannelPrivate,
+    RPCError,
     SessionPasswordNeeded,
     VolumeLocNotFound,
 )
@@ -352,6 +353,27 @@ def __init__(
     async def __aenter__(self):
         return await self.start()
 
+    @property
+    def connection_health(self) -> dict:
+        """Snapshot of the main session's connection health.
+
+        Includes ``state``, ``is_connected``, ``dc_id``, ``reconnect_attempt``,
+        ``pending_requests``, ``last_received_at`` and ``last_error``.
+        This property performs no network requests. Use it to inspect transient
+        recovery independently of the client lifecycle flag ``is_connected``.
+        """
+        if self.session is None:
+            return {
+                "state": "stopped",
+                "is_connected": False,
+                "dc_id": None,
+                "reconnect_attempt": 0,
+                "pending_requests": 0,
+                "last_received_at": None,
+                "last_error": None,
+            }
+        return self.session.health
+
     async def __aexit__(self, *args):
         with contextlib.suppress(ConnectionError):
             await self.stop()
@@ -382,7 +404,12 @@ async def updates_watchdog(self):
                 if self.update_manager is not None:
                     self.update_manager.request_recovery()
                 else:
-                    await self.invoke(raw.functions.updates.GetState())
+                    try:
+                        await self.invoke(raw.functions.updates.GetState())
+                    except (OSError, asyncio.TimeoutError, RPCError):
+                        # An idle period is not a transport failure. Session pings
+                        # own liveness checks and coordinate any necessary recovery.
+                        log.warning("Update watchdog probe failed", exc_info=True)
 
     async def authorize(self) -> User:
         if self.bot_token:
 
@@ -21,6 +21,7 @@
 
 import asyncio
 import logging
+import random
 from typing import TYPE_CHECKING
 
 from hydrogram.session.internals import DataCenter
@@ -56,16 +57,21 @@ def __init__(
         self.protocol: TCP | None = None
 
     async def connect(self) -> None:
-        for i in range(Connection.MAX_CONNECTION_ATTEMPTS):
+        for i in range(self.MAX_CONNECTION_ATTEMPTS):
             self.protocol = self.protocol_factory(ipv6=self.ipv6, proxy=self.proxy)
 
             try:
                 log.info("Connecting...")
                 await self.protocol.connect(self.address)
-            except OSError as e:
+            except (OSError, asyncio.TimeoutError) as e:
                 log.warning("Unable to connect due to network issues: %s", e)
                 await self.protocol.close()
-                await asyncio.sleep(1)
+                if i + 1 == self.MAX_CONNECTION_ATTEMPTS:
+                    raise ConnectionError("Transport connection attempts exhausted") from e
+                await asyncio.sleep(random.uniform(0.5, 1.0) * 2**i)
+            except BaseException:
+                await self.protocol.close()
+                raise
             else:
                 log.info(
                     "Connected! %s DC%s%s - IPv%s",
@@ -76,15 +82,19 @@ async def connect(self) -> None:
                 )
                 break
         else:
-            log.warning("Connection failed! Trying again...")
-            raise ConnectionError
+            raise ConnectionError("No transport connection attempts configured")
 
     async def close(self) -> None:
-        await self.protocol.close()
+        if self.protocol is not None:
+            await self.protocol.close()
         log.info("Disconnected")
 
     async def send(self, data: bytes) -> None:
+        if self.protocol is None:
+            raise ConnectionError("Transport is not connected")
         await self.protocol.send(data)
 
     async def recv(self) -> bytes | None:
+        if self.protocol is None:
+            return None
         return await self.protocol.recv()
@@ -61,7 +61,11 @@ def __init__(self, ipv6: bool, proxy: Proxy) -> None:
     @property
     def closed(self) -> bool:
         return (
-            self._closed or self.writer is None or self.writer.is_closing() or self.reader is None
+            self._closed
+            or self.writer is None
+            or self.writer.is_closing()
+            or self.reader is None
+            or self.reader.at_eof()
         )
 
     async def _connect_via_proxy(self, destination: tuple[str, int]) -> None:
@@ -93,11 +97,13 @@ async def _connect_via_proxy(self, destination: tuple[str, int]) -> None:
         )
         sock.settimeout(TCP.TIMEOUT)
 
-        await self.loop.sock_connect(sock=sock, address=destination)
-
-        sock.setblocking(False)
-
-        self.reader, self.writer = await asyncio.open_connection(sock=sock)
+        try:
+            await self.loop.sock_connect(sock=sock, address=destination)
+            sock.setblocking(False)
+            self.reader, self.writer = await asyncio.open_connection(sock=sock)
+        except BaseException:
+            sock.close()
+            raise
 
     async def _connect_via_direct(self, destination: tuple[str, int]) -> None:
         host, port = destination
@@ -136,13 +142,12 @@ async def close(self) -> None:
             self._closed = True
 
     async def send(self, data: bytes) -> None:
-        if self.writer is None or self._closed:
-            raise OSError("Connection is closed")
-
         async with self.lock:
+            if self.closed:
+                raise OSError("Connection is closed")
             try:
                 self.writer.write(data)
-                await self.writer.drain()
+                await asyncio.wait_for(self.writer.drain(), self.TIMEOUT)
             except Exception as e:
                 log.info("Send exception: %s %s", type(e).__name__, e)
                 self._closed = True
@@ -157,7 +162,10 @@ async def recv(self, length: int = 0) -> bytes | None:
         while len(data) < length:
             try:
                 chunk = await asyncio.wait_for(self.reader.read(length - len(data)), TCP.TIMEOUT)
-            except (OSError, asyncio.TimeoutError):
+            except asyncio.TimeoutError:
+                # A quiet socket is normal. Session pings detect unresponsive peers.
+                continue
+            except OSError:
                 self._closed = True
                 return None
             else:
 
@@ -26,6 +26,20 @@
 from .rpc_error import UnknownError
 
 
+class ConnectionLostError(ConnectionError):
+    """The connection was lost; a sent request's delivery may be uncertain."""
+
+    def __init__(self, message: str, *, delivery_uncertain: bool = True) -> None:
+        self.delivery_uncertain = delivery_uncertain
+        super().__init__(message)
+
+
+class RequestTimeoutError(TimeoutError):
+    """A sent request timed out. It may have been processed by Telegram."""
+
+    delivery_uncertain: bool = True
+
+
 class BadMsgNotification(Exception):  # noqa: N818
     descriptions: ClassVar = {
         16: "The msg_id is too low, the client time has to be synchronized.",
@@ -74,8 +88,10 @@ def __init__(self, msg: str | None = None):
 __all__ = [
     "BadMsgNotification",
     "CDNFileHashMismatch",
+    "ConnectionLostError",
     "ListenerStopped",
     "ListenerTimeout",
+    "RequestTimeoutError",
     "SecurityCheckMismatch",
     "SecurityError",
     "UnknownError",
 
@@ -59,7 +59,8 @@ async def invoke(
                 The API Schema function filled with proper arguments.
 
             retries (``int``):
-                Number of retries.
+                Attempt limit for requests not yet sent and explicit flood waits.
+                Requests with uncertain delivery and server errors are not replayed.
 
             timeout (``float``):
                 Timeout in seconds.
@@ -72,6 +73,10 @@ async def invoke(
 
         Raises:
             RPCError: In case of a Telegram RPC error.
+            ConnectionLostError: If the transport fails. Check ``delivery_uncertain``
+                before deciding whether the request can be safely retried.
+            RequestTimeoutError: If Telegram does not respond in time. The request
+                may have been processed; check its outcome before retrying.
         """
         if not self.is_connected:
             raise ConnectionError("Client has not been started yet")
 
@@ -51,7 +51,12 @@ async def connect(
             await self.storage.test_mode(),
         )
 
-        await self.session.start()
+        try:
+            await self.session.start()
+        except BaseException:
+            await self.session.stop()
+            await self.storage.close()
+            raise
 
         self.is_connected = True
 
 
@@ -17,10 +17,12 @@
 #  You should have received a copy of the GNU Lesser General Public License
 #  along with Hydrogram.  If not, see <http://www.gnu.org/licenses/>.
 
+import asyncio
 import logging
 
 import hydrogram
 from hydrogram import raw
+from hydrogram.errors import RPCError
 
 log = logging.getLogger(__name__)
 
@@ -41,8 +43,13 @@ async def terminate(
             raise ConnectionError("Client is already terminated")
 
         if self.takeout_id:
-            await self.invoke(raw.functions.account.FinishTakeoutSession())
-            log.info("Takeout session %s finished", self.takeout_id)
+            try:
+                await self.invoke(raw.functions.account.FinishTakeoutSession())
+            except (OSError, asyncio.TimeoutError, RPCError):
+                log.warning("Could not finish takeout session during shutdown", exc_info=True)
+            else:
+                log.info("Takeout session %s finished", self.takeout_id)
+            self.takeout_id = None
 
         if self.update_manager is not None:
             await self.update_manager.stop()
@@ -58,7 +65,9 @@ async def terminate(
         self.updates_watchdog_event.set()
 
         if self.updates_watchdog_task is not None:
-            await self.updates_watchdog_task
+            self.updates_watchdog_task.cancel()
+            await asyncio.gather(self.updates_watchdog_task, return_exceptions=True)
+            self.updates_watchdog_task = None
 
         self.updates_watchdog_event.clear()
 
 
@@ -59,6 +59,14 @@ async def main():
         """
 
         async def do_it():
+            # Do not make worker shutdown wait for an in-flight reconnect budget.
+            # Pending RPCs must be released before waiting on their handlers.
+            if (
+                self.session is not None
+                and self.session.recovery_task is not None
+                and not self.session.recovery_task.done()
+            ):
+                await self.session.stop()
             await self.terminate()
             await self.disconnect()
 
 
@@ -66,7 +66,8 @@ def unpack(b: BytesIO):
     async def invoke(self, data: TLObject):
         data = self.pack(data)
         await self.connection.send(data)
-        response = BytesIO(await self.connection.recv())
+        # Authentication runs before the session's ping monitor is available.
+        response = BytesIO(await asyncio.wait_for(self.connection.recv(), 15))
 
         return self.unpack(response)