fix(relay): shorten receiver reconnect gaps and count role_busy refusals

docs/i18n-manifest.json

@@ -21,7 +21,7 @@
     "docs/download-metrics.md": "ded60e7b84d14ce779f8d639516b4e390358a06ef76b677dc9c1f997fb626a9f",
     "docs/mcp.md": "e99689b8e6a222babc9438be2b2aabee953c3e0b2f967c6582065569529c32da",
     "docs/multi-home-list.md": "f8d8e3adf727e4842ea7c4d45e8f6523b11d0513bf2509d4d1204645875e6489",
-    "docs/paired-devices.md": "0145495adfacc929544275000f3fb659a5adf909faafe200fbc47ca7961258a2",
+    "docs/paired-devices.md": "b766b51086f05dda3408005c27a3b02e4463a62dfb2c232422a4d3ae0860c94b",
     "docs/relay-auth.md": "e860747a5ef3b87157500d4334fe1d10852b8d65c67a4042d81f7ee5e11ea184",
     "docs/relay-development-plan.md": "0abf750b36275ea4e467a0020e560277f08689f2d574aed7674133705383b1ff",
     "docs/relay-edge-policy.md": "c1c7f7a1c4303943c78ca334dab4034f66c0c272e40c007d07c3da2a496155eb",

docs/ja/paired-devices.md

@@ -12,7 +12,9 @@
 
 WebSocket 切断の `closeSource` はコードの受信・送信方向を区別します。Relay のカウンターは受信側先着とクライアント先着のルームを区別します。`receiverWsAgeMs` は待機時間であり、接続の健全性を証明しません。ペアリングと受信側の `attach_received`・ピア TLS イベントの時刻を照合してください。
 
-`stream_close.closeCode` はリモートから受信した切断コードです。`1006` は切断フレームを受信しなかったことを意味し、レグの停止を示す可能性があります。`peer_closed` は相手レグの終了後に Relay がこのレグを閉じた場合、`remote_going_away` は Relay の切断要求なしにリモート側が 1001 を送った場合です。
+`stream_close.closeCode` はリモートから受信した切断コードです。`1006` は切断フレームを受信しなかったことを意味し、レグの停止を示す可能性があります。`peer_closed` は相手レグの終了後に Relay がこのレグを閉じた場合、`remote_going_away` は Relay の切断要求なしにリモート側が 1001 を送った場合です。`receiverRoleBusy`・`clientRoleBusy` は同じルームで拒否された重複レグの数です。受信側の値が増える場合、受信側の再接続中に Relay が古いレグを保持している可能性があります。
+
+1 秒以上開いていた待機ルームが Relay により正常に閉じられた場合、受信側はバックオフを増やさず 0.5 秒後に再接続します。その他の失敗と 1 秒以内に閉じたルームは、引き続き最大 5 秒まで指数バックオフします。Relay レグは両端で 10 秒の WebSocket ping 間隔・タイムアウトを使うため、停止したレグを約 40 秒ではなく約 20 秒で検出します。これは再接続の空白を減らすだけで、停止したネットワーク経路を修復するものではありません。
 
 ## インストール
 

docs/ko/paired-devices.md

@@ -12,7 +12,9 @@
 
 WebSocket 종료의 `closeSource`는 코드가 수신됐는지 송신됐는지 구분합니다. Relay 지표는 수신자 선착·클라이언트 선착 방을 구분합니다. `receiverWsAgeMs`는 수신자 대기 시간이지 연결 생존의 증거가 아닙니다. 방 페어링과 수신자의 `attach_received`·피어 TLS 이벤트 시각을 대조해야 합니다.
 
-`stream_close.closeCode`는 원격에서 수신한 종료 코드입니다. `1006`은 종료 프레임을 받지 못했다는 뜻이며 레그 정체 가능성을 시사합니다. `peer_closed`는 상대 레그가 끝나 Relay가 이 레그를 닫은 경우, `remote_going_away`는 Relay의 닫기 요청 없이 원격 엔드포인트가 1001을 보낸 경우입니다.
+`stream_close.closeCode`는 원격에서 수신한 종료 코드입니다. `1006`은 종료 프레임을 받지 못했다는 뜻이며 레그 정체 가능성을 시사합니다. `peer_closed`는 상대 레그가 끝나 Relay가 이 레그를 닫은 경우, `remote_going_away`는 Relay의 닫기 요청 없이 원격 엔드포인트가 1001을 보낸 경우입니다. `receiverRoleBusy`·`clientRoleBusy`는 같은 방에서 거절된 중복 레그 수이며, 수신자 값이 늘면 수신자가 재연결하는 동안 Relay가 이전 레그를 아직 붙잡고 있을 수 있습니다.
+
+1초 이상 열려 있던 대기 방이 Relay에 의해 정상 종료되면 수신자는 백오프를 늘리지 않고 0.5초 뒤 다시 연결합니다. 그 밖의 실패와 1초 안에 닫힌 방은 계속 최대 5초까지 지수 백오프합니다. Relay 레그는 양쪽 모두 10초 WebSocket ping 간격·제한을 사용하므로 정체된 레그를 약 40초가 아닌 약 20초 안에 감지합니다. 이는 재연결 공백을 줄일 뿐 정체된 네트워크 경로를 고치지는 않습니다.
 
 ## 설치
 

docs/paired-devices.md

@@ -48,6 +48,17 @@ receiver's `attach_received` and peer-TLS events before inferring a stale leg.
 `1006` means no close frame was received (a possible stalled leg).
 `peer_closed` means the Relay closed this leg after its counterpart ended;
 `remote_going_away` means the remote endpoint sent 1001 without a Relay-initiated close.
+`receiverRoleBusy` and `clientRoleBusy` count refused duplicate legs for the same
+room; a rising receiver count can mean the Relay still holds an old leg while the
+receiver reconnects.
+
+A receiver whose waiting room stayed open for at least one second and was then
+closed normally by the Relay reconnects after 0.5 seconds instead of escalating
+its backoff. Other failures, and rooms closed within one second, still back off
+exponentially up to 5 seconds. Relay legs use a 10-second WebSocket ping interval
+and timeout on both endpoints, so a stalled leg is detected in about 20 seconds
+rather than about 40. These reduce reconnect gaps; they do not repair a stalled
+network path.
 
 ## Install
 

docs/zh-CN/paired-devices.md

@@ -15,7 +15,9 @@ NAT 穿透、WireGuard 隧道或自动公开服务。
 
 WebSocket 关闭的 `closeSource` 区分代码来自接收还是发送。Relay 计数器区分接收端先到和客户端先到的房间。`receiverWsAgeMs` 只表示等待时长，不能证明连接健康；需要对照配对时间与接收端的 `attach_received`、对端 TLS 事件。
 
-`stream_close.closeCode` 是从远端收到的关闭代码。`1006` 表示未收到关闭帧，可能提示连接停滞。`peer_closed` 表示另一条连接结束后 Relay 关闭了本连接；`remote_going_away` 表示远端在 Relay 未发起关闭时自行发送了 1001 代码。
+`stream_close.closeCode` 是从远端收到的关闭代码。`1006` 表示未收到关闭帧，可能提示连接停滞。`peer_closed` 表示另一条连接结束后 Relay 关闭了本连接；`remote_going_away` 表示远端在 Relay 未发起关闭时自行发送了 1001 代码。`receiverRoleBusy`、`clientRoleBusy` 统计同一房间中被拒绝的重复连接；接收端计数上升可能意味着接收端重连时 Relay 仍持有旧连接。
+
+若等待房间已开启至少 1 秒后被 Relay 正常关闭，接收端会在 0.5 秒后重连，而不会增加退避时间。其他失败以及 1 秒内关闭的房间仍按指数退避，最长 5 秒。Relay 连接两端均使用 10 秒的 WebSocket ping 间隔与超时，因此停滞连接约 20 秒即可检测到，而不是约 40 秒。这些改动只缩短重连空档，并不能修复停滞的网络路径。
 
 ## 安装
 

session_peer_relay/app.py

@@ -206,6 +206,13 @@ def observed(event, *, elapsed_ms):
                 idle = (detail.stage == 'relay_attach' and detail.close_code == 1000
                         and detail.close_source == 'received'
                         and websocket_opened is not None and 50 <= now-websocket_opened <= 70)
+                # A normal Relay close of a room that stayed open is an expected
+                # idle expiry, not a failure: reconnect promptly instead of
+                # escalating the backoff. Rooms closed within a second keep the
+                # exponential backoff so a misbehaving Relay cannot cause a hot loop.
+                expired = (detail.stage == 'relay_attach' and detail.close_code == 1000
+                           and detail.close_source == 'received'
+                           and websocket_opened is not None and now-websocket_opened >= 1)
                 self.lifecycle_event('websocket_closed' if websocket_opened is not None else 'setup_failed',
                     reason='idle_expiry_like' if idle else detail.diagnostic()['reason'],
                     stage=detail.stage, closeCode=detail.close_code, closeSource=detail.close_source,
@@ -217,8 +224,10 @@ def observed(event, *, elapsed_ms):
                     self.relay_failure_last_logged = now
                     logging.getLogger(__name__).warning('relay_connection_failed %s',
                         json.dumps(detail.diagnostic(), sort_keys=True))
+                if expired:
+                    delay = .5
                 await asyncio.sleep(delay)
-                delay = min(5, delay*2)
+                delay = .5 if expired else min(5, delay*2)
 
     async def close(self):
         tasks = list(self.tasks)

session_peer_relay/relay.py

@@ -80,6 +80,8 @@ def __init__(self, accounts, *, capture=None, control=None, limits=None, diagnos
             'receiverIdleExpired': 0,
             'clientWaitExpired': 0,
             'roomsClosedBeforeAttach': 0,
+            'receiverRoleBusy': 0,
+            'clientRoleBusy': 0,
         }
 
     def diagnostic_event(self, event, **values):
@@ -215,6 +217,10 @@ async def watch_authorization():
         try:
             if account['role'] in slot['members']:
                 close_reason = 'role_busy'
+                # Counted per role so a leg the Relay still holds while its
+                # endpoint reconnects is visible in metrics.
+                self.counters['receiverRoleBusy' if account['role'] == 'receiver' else 'clientRoleBusy'] += 1
+                self.diagnostic_event('role_busy', roomId=slot['id'], role=account['role'])
                 await ws.close(1013, 'role_busy')
                 return
             slot['members'][account['role']] = ws

session_peer_relay/wire.py

@@ -124,7 +124,10 @@ def observe(event, started):
         ws = await PinnedConnect(url, additional_headers={'Cookie': cookie}, compression=None,
                            user_agent_header='session-peer/0.9-relay',
                            max_size=MAX_FRAME, max_queue=4, write_limit=32768,
-                           open_timeout=TIMEOUT, close_timeout=2)
+                           open_timeout=TIMEOUT, close_timeout=2,
+                           # Match the Relay's 10/10 keepalive so a stalled leg is
+                           # detected within about 20 s instead of about 40 s.
+                           ping_interval=10, ping_timeout=10)
     except InvalidStatus as exc:
         raise TransportFailure('relay_websocket', 'http_rejected', http_status=exc.response.status_code) from None
     except Exception as exc:

tests/relay/test_receiver_reconnect.py

@@ -0,0 +1,109 @@
+"""Receiver reconnect behavior (#163): idle-expiry backoff, keepalive, role_busy metrics."""
+import asyncio
+import time
+import unittest
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, Mock, patch
+
+from tests.relay import test_native_relay as platform_guard  # skips without relay extras
+from session_peer_relay import app, wire
+from session_peer_relay.transport_errors import TransportFailure
+
+
+class ReceiverIdleExpiryBackoff(unittest.IsolatedAsyncioTestCase):
+    async def sleeps_for(self, rooms):
+        """rooms: (open seconds, close code, close source) per attempt."""
+        receiver = app.Receiver(Mock(), {'targets': {'worker': {'agent': 'claude', 'target': 'worker'}},
+                                         'peers': {}})
+        clock, sleeps, calls = [0.0], [], [0]
+
+        async def connect(url, credential, *, attach_timeout, on_event):
+            if calls[0] == len(rooms):
+                raise asyncio.CancelledError()
+            age, code, source = rooms[calls[0]]
+            calls[0] += 1
+            on_event('websocket_open', elapsed_ms=1)
+            clock[0] += age
+            raise TransportFailure('relay_attach', 'connection_closed',
+                                   close_code=code, close_source=source)
+
+        async def sleep(seconds):
+            sleeps.append(seconds)
+            clock[0] += seconds
+
+        with patch.object(app, 'relay_stream', side_effect=connect), \
+             patch.object(app, 'time', SimpleNamespace(monotonic=lambda: clock[0], time=time.time)), \
+             patch.object(app.asyncio, 'sleep', new=AsyncMock(side_effect=sleep)):
+            with self.assertRaises(asyncio.CancelledError):
+                await receiver.relay_listener('wss://relay.example.test/v1/connect', 'SECRET')
+        return sleeps
+
+    async def test_expected_idle_expiry_reconnects_without_escalation(self):
+        self.assertEqual(await self.sleeps_for([(60, 1000, 'received')] * 6), [.5] * 6)
+
+    async def test_short_rooms_keep_exponential_backoff(self):
+        self.assertEqual(await self.sleeps_for([(.2, 1000, 'received')] * 6), [.5, 1, 2, 4, 5, 5])
+
+    async def test_locally_sent_or_abnormal_close_keeps_backoff(self):
+        rooms = [(60, 1000, 'sent'), (60, 1011, 'sent'), (60, 1001, 'received')]
+        self.assertEqual(await self.sleeps_for(rooms), [.5, 1, 2])
+
+    async def test_idle_expiry_after_failures_resets_to_prompt_reconnect(self):
+        rooms = [(.2, 1000, 'received')] * 3 + [(60, 1000, 'received')] * 2
+        self.assertEqual(await self.sleeps_for(rooms), [.5, 1, 2, .5, .5])
+
+
+class RelayLegKeepalive(unittest.IsolatedAsyncioTestCase):
+    async def test_relay_legs_use_ten_second_keepalive(self):
+        ws = SimpleNamespace(recv=AsyncMock(return_value='{"relayAttached": true}'), close=AsyncMock())
+        with patch.object(wire, 'admission', return_value='session_peer=SECRET'), \
+             patch.object(wire, 'PinnedConnect', new=AsyncMock(return_value=ws)) as connect:
+            await wire.relay_stream('ws://127.0.0.1:2/v1/connect', 'SECRET')
+        kwargs = connect.await_args.kwargs
+        self.assertEqual((kwargs['ping_interval'], kwargs['ping_timeout']), (10, 10))
+
+
+
+class RelayRoleBusyMetrics(unittest.IsolatedAsyncioTestCase):
+    async def test_duplicate_role_is_refused_and_counted_per_role(self):
+        from session_peer_relay.relay import Relay
+
+        class Socket:
+            def __init__(self, role):
+                self.lab_account = {'room': 'SECRET-ROOM', 'role': role}
+                self.lab_expiry = time.monotonic() + 10
+                self.closed = asyncio.Event()
+                self.close_code = None
+
+            async def wait_closed(self):
+                await self.closed.wait()
+
+            async def close(self, code=1000, reason=''):
+                if not self.closed.is_set():
+                    self.close_code = code
+                self.closed.set()
+
+        relay = Relay([], diagnostic_events=True)
+        relay.receiver_wait = relay.client_wait = .3
+        with self.assertLogs('session_peer_relay.relay', level='WARNING') as logs:
+            for role in ('receiver', 'client'):
+                first, second = Socket(role), Socket(role)
+                waiting = asyncio.create_task(relay.handler(first))
+                async with asyncio.timeout(1):
+                    while relay.metrics()['current']['waitingRooms'] != 1:
+                        await asyncio.sleep(.002)
+                await relay.handler(second)
+                self.assertEqual(second.close_code, 1013)
+                self.assertIsNone(first.close_code)
+                await first.close()
+                await waiting
+        for role in ('receiver', 'client'):
+            self.assertTrue(any('"role_busy"' in row and f'"role": "{role}"' in row
+                                for row in logs.output))
+        self.assertNotIn('SECRET-ROOM', '\n'.join(logs.output))
+        counters = relay.metrics()['counters']
+        self.assertEqual((counters['receiverRoleBusy'], counters['clientRoleBusy']), (1, 1))
+
+
+if __name__ == '__main__':
+    unittest.main()