@@ -52,8 +52,13 @@ Library callers that omit `message_spacing` still get `DEFAULT_MESSAGE_SPACING`
 **timeout**, `Sunsynk.read_holding_registers()` logs, increments **`timeouts`**, and retries up to
 **`read_attempts`** (add-on **`READ_ATTEMPTS`**, default **3**) before raising **`ExceptionGroup`**.
 **Serial** links also call `ModbusConnection.disconnect()` (not `close()`) so the next FC03
-reconnects with an empty receive buffer. TCP, UDP, and Solarman do not flush on timeout. Other read
-errors still flush. tmodbus already disconnects on desync (bad function code / header mismatch).
+reconnects with an empty receive buffer. TCP and UDP do not flush on a single timeout (the MBAP
+transaction ID already discards a late reply), but after **`dead_link_timeouts`** (default **6**)
+consecutive timeouts, across reads, the connection is recycled: some RS485 gateways stop serving one
+TCP session while still accepting new ones, and without this the socket stays dead until the kernel
+gives up (`tcp_retries2`, ~16 min). Any successful read resets the count. Solarman does not flush on
+timeout. Other read errors still flush. tmodbus already disconnects on desync (bad function code /
+header mismatch).
 
 ---
 
 
@@ -2,6 +2,11 @@
 
 <https://github.com/kellerza/sunsynk/commits/>
 
+## Unreleased
+
+- TCP/UDP: recycle the Modbus connection after 6 consecutive read timeouts, so a gateway session
+  that stopped answering no longer stalls polling until the kernel drops the socket (~16 min).
+
 ## Release 1.3.0
 
 All items below are changes since add-on **1.2.0**.
 
@@ -82,6 +82,18 @@ class Sunsynk:
     read_sensors_batch_size: int = 20
     allow_gap: int = 2
     timeouts: int = 0
+    dead_link_timeouts: int = 6
+    """Consecutive non-serial read timeouts before recycling the link (0 disables).
+
+    A gateway can stop serving one TCP session while still accepting new ones: every
+    request on the old socket then times out until the kernel gives up (``tcp_retries2``,
+    ~16 min). A single late reply needs no flush on TCP (the transaction ID discards it).
+    Not applied without a ModbusConnection (Solarman).
+    """
+    # ponytail: counted per Sunsynk instance (per MODBUS_ID). Inverters on one port share
+    # the connection, so an unresponsive unit can recycle a healthy one's link; upgrade
+    # path: keep the count on the shared connection.
+    _consecutive_timeouts: int = field(default=0, init=False, repr=False)
 
     @property
     def retrying_unit(self) -> HoldingUnit:
@@ -189,9 +201,10 @@ async def read_holding_registers(self, start: int, length: int) -> Sequence[int]
         errs: list[Exception] = []
         for attempt in range(self.read_attempts):
             try:
-                return await self.unit.read_holding_registers(start, length)
+                regs = await self.unit.read_holding_registers(start, length)
             except ModbusTimeoutError as err:
                 self.timeouts += 1
+                self._consecutive_timeouts += 1
                 errs.append(err)
                 _LOG.error(
                     "Read register %s x%s: %s [attempt %s/%s]",
@@ -205,6 +218,18 @@ async def read_holding_registers(self, start: int, length: int) -> Sequence[int]
                     self.connection._params, ModbusSerialParams
                 ):
                     await self._flush_modbus_connection(reason=type(err).__name__)
+                elif (
+                    self.connection is not None
+                    and self.dead_link_timeouts
+                    and self._consecutive_timeouts >= self.dead_link_timeouts
+                ):
+                    _LOG.warning(
+                        "%s consecutive timeouts: recycling the Modbus connection (%s)",
+                        self._consecutive_timeouts,
+                        self.port,
+                    )
+                    self._consecutive_timeouts = 0
+                    await self._flush_modbus_connection(reason="dead link")
             except ModbusProtocolError as err:
                 errs.append(err)
                 _LOG.error(
@@ -225,6 +250,9 @@ async def read_holding_registers(self, start: int, length: int) -> Sequence[int]
                     self.read_attempts,
                 )
                 await self._flush_modbus_connection(reason=type(err).__name__)
+            else:
+                self._consecutive_timeouts = 0
+                return regs
 
         raise ExceptionGroup(f"Failed to read {length} registers at {start}", errs)
 
 
@@ -178,6 +178,64 @@ async def test_read_holding_registers_timeout_tcp_does_not_flush(
     conn.disconnect.assert_not_awaited()
 
 
+async def test_read_holding_registers_tcp_dead_link_recycles(
+    state: InverterState,
+) -> None:
+    """Consecutive TCP timeouts across reads recycle the connection once."""
+    unit = MagicMock()
+    unit.read_holding_registers = AsyncMock(side_effect=ModbusTimeoutError)
+    conn = MagicMock()
+    conn.connected = True
+    conn.disconnect = AsyncMock()
+    conn._params = ModbusTcpParams(host="1.2.3.4", port=502)
+    ss = Sunsynk(
+        unit=unit, state=state, connection=conn, read_attempts=2, dead_link_timeouts=3
+    )
+    for _ in range(2):
+        with pytest.raises(ExceptionGroup, match="Failed to read"):
+            await ss.read_holding_registers(1, 1)
+    assert ss.timeouts == 4
+    conn.disconnect.assert_awaited_once()
+
+
+async def test_read_holding_registers_tcp_success_resets_dead_link_count(
+    state: InverterState,
+) -> None:
+    """Isolated TCP timeouts, each recovered by a retry, never recycle the link."""
+    unit = MagicMock()
+    unit.read_holding_registers = AsyncMock(
+        side_effect=[ModbusTimeoutError, [1], ModbusTimeoutError, [2]] * 3
+    )
+    conn = MagicMock()
+    conn.connected = True
+    conn.disconnect = AsyncMock()
+    conn._params = ModbusTcpParams(host="1.2.3.4", port=502)
+    ss = Sunsynk(
+        unit=unit, state=state, connection=conn, read_attempts=2, dead_link_timeouts=2
+    )
+    for _ in range(6):
+        assert await ss.read_holding_registers(1, 1) in ([1], [2])
+    assert ss.timeouts == 6
+    conn.disconnect.assert_not_awaited()
+
+
+async def test_read_holding_registers_tcp_dead_link_disabled(
+    state: InverterState,
+) -> None:
+    """``dead_link_timeouts=0`` keeps the pre-existing TCP behaviour."""
+    unit = MagicMock()
+    unit.read_holding_registers = AsyncMock(side_effect=ModbusTimeoutError)
+    conn = MagicMock()
+    conn.connected = True
+    conn.disconnect = AsyncMock()
+    conn._params = ModbusTcpParams(host="1.2.3.4", port=502)
+    ss = Sunsynk(unit=unit, state=state, connection=conn, dead_link_timeouts=0)
+    for _ in range(5):
+        with pytest.raises(ExceptionGroup, match="Failed to read"):
+            await ss.read_holding_registers(1, 1)
+    conn.disconnect.assert_not_awaited()
+
+
 async def test_read_holding_registers_timeout_without_connection(
     state: InverterState,
 ) -> None:
@@ -190,6 +248,19 @@ async def test_read_holding_registers_timeout_without_connection(
     assert ss.timeouts == ss.read_attempts
 
 
+async def test_read_holding_registers_dead_link_without_connection(
+    state: InverterState, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Without a ModbusConnection there is nothing to recycle: no false warning."""
+    unit = MagicMock()
+    unit.read_holding_registers = AsyncMock(side_effect=ModbusTimeoutError)
+    ss = Sunsynk(unit=unit, state=state, read_attempts=2, dead_link_timeouts=2)
+    for _ in range(3):
+        with pytest.raises(ExceptionGroup, match="Failed to read"):
+            await ss.read_holding_registers(1, 1)
+    assert "recycling the Modbus connection" not in caplog.text
+
+
 async def test_read_holding_registers_timeout_then_success(
     state: InverterState,
 ) -> None: