feat(relay): record per-leg frame progress for stall triage

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": "f20c44e99237324a3a3264e7756ce8896f02551a3b30f79210193ffd96cf0d56",
+    "docs/paired-devices.md": "818f2e5ea56bdfbbc2eb847ebec53b293df104471e4ab20f4059799692e6e18d",
     "docs/relay-auth.md": "4c5d3d6d10c0d607c976c10617c51ed4b21344e291b0fabf54c4960040846df2",
     "docs/relay-development-plan.md": "0abf750b36275ea4e467a0020e560277f08689f2d574aed7674133705383b1ff",
     "docs/relay-edge-policy.md": "c1c7f7a1c4303943c78ca334dab4034f66c0c272e40c007d07c3da2a496155eb",

docs/ja/paired-devices.md

@@ -16,6 +16,8 @@ WebSocket 切断の `closeSource` はコードの受信・送信方向を区別
 
 明示的に有効にした Relay と受信側のライフサイクルイベントには `eventTimeUtcMs`（UTC Unix ミリ秒）が含まれます。クライアントは Relay 接続の試行ごとにランダムな UUIDv4 `attemptId` を生成し、成功結果または試行ごとの失敗メタデータに記録します。新しい Relay はこの ID をペアリング済みルームイベントに記録し、attach 通知で新しい受信側へ渡します。受信側の `attach_received` と peer-TLS イベントにも同じ ID が記録されます。この ID は ID 情報、ルーム名、リクエスト ID、メッセージ内容から導出されず、認可や再送の安全性も変えません。失敗を帰属させるには、完全一致する `attemptId` が Relay のルーム一つと受信側レグ一つだけに対応することを確認し、時計のずれを考慮して UTC 時刻を比較してください。イベントが欠ける、重複する、または旧バージョンのレグの場合は、時刻や集計カウンターから推測せず `unattributed` と記録してください。
 
+公開経路の停止が再発した場合、明示的に有効にした Relay の `stream_close` イベントで、各レグの `ingressFrames`/`ingressBytes`、`egressCompletedFrames`/`egressCompletedBytes`、最初と最後の UTC フレーム時刻を比較できます。これは暗号化された WebSocket フレームの数であり、メッセージ数ではありません。送信完了は Relay のソケット send が戻ったことだけを示し、Cloudflare、受信側、TLS、アプリケーションがバイトを消費した証明ではありません。同じルームと `attemptId` の両方のロールを比較してから受信側イベントと照合してください。イベントが欠けるか一致が曖昧なら `unattributed` のままにします。カウンターだけで Cloudflare、OCI、中間ホップの責任は特定できません。メタデータの時刻とフレームサイズからも活動が分かるため、opt-in ログは非公開で期間を限定して保持してください。
+
 1 秒以上開いていた待機ルームが Relay により正常に閉じられた場合、受信側はバックオフを増やさず 0.5 秒後に再接続します。その他の失敗と 1 秒以内に閉じたルームは、引き続き最大 5 秒まで指数バックオフします。Relay レグは両端で 10 秒の WebSocket ping 間隔・タイムアウトを使うため、停止したレグを約 40 秒ではなく約 20 秒で検出します。これは再接続の空白を減らすだけで、停止したネットワーク経路を修復するものではありません。
 
 ## インストール

docs/ko/paired-devices.md

@@ -16,6 +16,8 @@ WebSocket 종료의 `closeSource`는 코드가 수신됐는지 송신됐는지 
 
 명시적으로 켠 Relay·수신자 수명주기 이벤트에는 `eventTimeUtcMs`(UTC Unix 밀리초)가 포함됩니다. 클라이언트는 Relay 연결 시도마다 임의 UUIDv4 `attemptId`를 새로 만들고 성공 결과 또는 시도별 실패 메타데이터에 기록합니다. 새 Relay는 이 ID를 페어링된 방 이벤트에 기록하고 새 수신자에게 attach 알림으로 전달합니다. 수신자의 `attach_received`·peer-TLS 이벤트에도 같은 ID가 남습니다. 이 ID는 신원·방 이름·요청 ID·메시지 내용에서 파생되지 않으며 인가나 재전송 안전성을 바꾸지 않습니다. 실패를 귀속하려면 정확한 `attemptId`가 Relay 방 하나와 수신자 레그 하나에만 대응하는지 확인하고 시계 오차를 고려해 UTC 시각을 대조하십시오. 이벤트가 없거나 중복되거나 구버전 레그라면 시각·집계값만으로 추정하지 말고 `unattributed`로 기록하십시오.
 
+공개 경로 정체가 재발하면 명시적으로 켠 Relay의 `stream_close` 이벤트에서 각 레그의 `ingressFrames`/`ingressBytes`, `egressCompletedFrames`/`egressCompletedBytes` 및 첫·마지막 UTC 프레임 시각을 비교할 수 있습니다. 이는 암호화된 WebSocket 프레임 수이지 메시지 수가 아닙니다. 송신 완료는 Relay의 소켓 send가 반환됐다는 뜻일 뿐 Cloudflare·수신자·TLS·애플리케이션이 바이트를 소비했다는 증거가 아닙니다. 같은 방·`attemptId` 아래 두 역할을 비교한 뒤 수신자 이벤트와 대조하십시오. 이벤트가 없거나 매칭이 모호하면 `unattributed`로 유지합니다. 카운터만으로 Cloudflare·OCI·중간 홉의 귀책을 정할 수 없습니다. 메타데이터 시각과 프레임 크기만으로도 활동이 드러날 수 있으므로 opt-in 로그를 비공개·한정 보존하십시오.
+
 1초 이상 열려 있던 대기 방이 Relay에 의해 정상 종료되면 수신자는 백오프를 늘리지 않고 0.5초 뒤 다시 연결합니다. 그 밖의 실패와 1초 안에 닫힌 방은 계속 최대 5초까지 지수 백오프합니다. Relay 레그는 양쪽 모두 10초 WebSocket ping 간격·제한을 사용하므로 정체된 레그를 약 40초가 아닌 약 20초 안에 감지합니다. 이는 재연결 공백을 줄일 뿐 정체된 네트워크 경로를 고치지는 않습니다.
 
 ## 설치

docs/paired-devices.md

@@ -64,6 +64,17 @@ receiver leg, then compare their UTC times with clock skew in mind. If any leg
 is missing, duplicated, or running an older version, record the attempt as
 `unattributed` instead of inferring a match from timing or aggregate counters.
 
+For a recurring public-path stall, an opt-in Relay `stream_close` event also
+reports each leg's `ingressFrames`/`ingressBytes` and
+`egressCompletedFrames`/`egressCompletedBytes`, with first and last UTC frame
+times. These count opaque encrypted WebSocket frames, not messages. An egress
+completion means the Relay's socket send returned; it does **not** prove
+Cloudflare, the receiver, TLS, or the application consumed those bytes. Compare
+both roles under the same room and `attemptId`, then the receiver events. Missing
+events or ambiguous matches remain `unattributed`; counters alone cannot assign
+fault to Cloudflare, OCI, or an intermediate hop. Keep opt-in logs private and
+bounded because even metadata timestamps and frame sizes reveal activity.
+
 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

docs/zh-CN/paired-devices.md

@@ -19,6 +19,8 @@ WebSocket 关闭的 `closeSource` 区分代码来自接收还是发送。Relay 
 
 显式启用的 Relay 和接收端生命周期事件包含 `eventTimeUtcMs`（UTC Unix 毫秒）。客户端每次尝试建立 Relay 连接都会生成新的随机 UUIDv4 `attemptId`，并在成功结果或逐次失败元数据中记录。新版 Relay 在已配对房间事件中记录该 ID，并通过 attach 通知传给新版接收端；接收端的 `attach_received` 和 peer-TLS 事件也记录同一 ID。该 ID 不由身份、房间名、请求 ID 或消息内容派生，也不会改变授权或重试安全性。归因失败时，应确认完全相同的 `attemptId` 只对应一个 Relay 房间和一条接收端连接，并在比较 UTC 时间时考虑时钟偏差。若事件缺失、重复或某条连接使用旧版本，不要仅凭时间或汇总计数推断，应记为 `unattributed`。
 
+如果公开路径停滞再次出现，可在显式启用的 Relay `stream_close` 事件中比较每条连接的 `ingressFrames`/`ingressBytes`、`egressCompletedFrames`/`egressCompletedBytes` 以及首末帧 UTC 时间。这些是加密 WebSocket 帧计数，不是消息数。发送完成只表示 Relay 的套接字 send 已返回，不能证明 Cloudflare、接收端、TLS 或应用程序收到了这些字节。先比较同一房间和 `attemptId` 下的两个角色，再对照接收端事件。事件缺失或匹配模糊时仍应记为 `unattributed`。仅凭计数无法判定 Cloudflare、OCI 或中间链路的责任。元数据时间和帧大小也会透露活动信息，因此应将 opt-in 日志私密且限期保存。
+
 若等待房间已开启至少 1 秒后被 Relay 正常关闭，接收端会在 0.5 秒后重连，而不会增加退避时间。其他失败以及 1 秒内关闭的房间仍按指数退避，最长 5 秒。Relay 连接两端均使用 10 秒的 WebSocket ping 间隔与超时，因此停滞连接约 20 秒即可检测到，而不是约 40 秒。这些改动只缩短重连空档，并不能修复停滞的网络路径。
 
 ## 安装

session_peer_relay/relay.py

@@ -208,6 +208,9 @@ async def handler(self, ws):
         paired = False
         attach_sent = False
         close_reason = 'socket_closed'
+        ingress_frames = ingress_bytes = egress_completed_frames = egress_completed_bytes = 0
+        first_ingress_utc_ms = last_ingress_utc_ms = None
+        first_egress_completed_utc_ms = last_egress_completed_utc_ms = None
         if self.control:
             async def watch_authorization():
                 while True:
@@ -275,16 +278,21 @@ async def watch_authorization():
                                   attemptId=slot['attemptId'])
             attach_sent = True
             close_reason = 'attached_closed'
-            forwarded_bytes = 0
             async with asyncio.timeout(max(0, ws.lab_expiry-time.monotonic())):
                 async for data in ws:
                     if not isinstance(data, bytes):
                         await ws.close(1008, 'binary_only')
                         break
+                    ingress_frames += 1
+                    ingress_bytes += len(data)
+                    if self.diagnostic_events:
+                        now_utc_ms = int(time.time()*1000)
+                        if first_ingress_utc_ms is None:
+                            first_ingress_utc_ms = now_utc_ms
+                        last_ingress_utc_ms = now_utc_ms
                     self.bytes += len(data)
-                    forwarded_bytes += len(data)
                     self.forwarded_frames += 1
-                    if forwarded_bytes > self.limits.connection_byte_budget:
+                    if ingress_bytes > self.limits.connection_byte_budget:
                         self.counters['byteBudgetClosed'] += 1
                         await ws.close(1013, 'byte_budget')
                         break
@@ -295,6 +303,13 @@ async def watch_authorization():
                     except TimeoutError:
                         close_reason = 'forward_send_timeout'
                         raise
+                    egress_completed_frames += 1
+                    egress_completed_bytes += len(data)
+                    if self.diagnostic_events:
+                        now_utc_ms = int(time.time()*1000)
+                        if first_egress_completed_utc_ms is None:
+                            first_egress_completed_utc_ms = now_utc_ms
+                        last_egress_completed_utc_ms = now_utc_ms
             if close_reason == 'attached_closed':
                 if account['role'] in slot['closedByRelay']:
                     close_reason = 'peer_closed'
@@ -332,7 +347,14 @@ async def watch_authorization():
                     self.diagnostic_event('stream_close', roomId=slot['id'], role=account['role'],
                         reason=close_reason, attachSent=attach_sent,
                         closeCode=code if type(code) is int and 1000 <= code <= 4999 else None,
-                        attemptId=slot['attemptId'])
+                        attemptId=slot['attemptId'],
+                        ingressFrames=ingress_frames, ingressBytes=ingress_bytes,
+                        egressCompletedFrames=egress_completed_frames,
+                        egressCompletedBytes=egress_completed_bytes,
+                        firstIngressUtcMs=first_ingress_utc_ms,
+                        lastIngressUtcMs=last_ingress_utc_ms,
+                        firstEgressCompletedUtcMs=first_egress_completed_utc_ms,
+                        lastEgressCompletedUtcMs=last_egress_completed_utc_ms)
 
     async def start(self, host, port):
         return await serve(self.handler, host, port, process_request=self.process_request,

tests/relay/test_relay_diagnostics.py

@@ -192,6 +192,38 @@ async def test_stalled_receiver_consumes_room_then_next_client_times_out(self):
         self.assertTrue(any('"reason": "attach_notify_failed"' in row for row in logs.output))
         self.assertNotIn('SECRET-ROOM', '\n'.join(logs.output))
 
+    async def test_paired_stream_records_only_per_leg_frame_metadata(self):
+        relay = Relay([], diagnostic_events=True)
+        receiver_payload = b'receiver-ciphertext-secret'
+        client_payload = b'client-ciphertext-secret'
+        receiver = FakeWebSocket('receiver', frames=[receiver_payload])
+        client = FakeWebSocket('client', frames=[client_payload])
+        tasks = []
+        with self.assertLogs('session_peer_relay.relay', level='WARNING') as logs:
+            try:
+                tasks.append(asyncio.create_task(relay.handler(receiver)))
+                await self.wait_for(lambda: relay.metrics()['current']['waitingRooms'] == 1)
+                tasks.append(asyncio.create_task(relay.handler(client)))
+                await self.wait_for(lambda: client_payload in receiver.sent
+                                    and receiver_payload in client.sent)
+            finally:
+                await receiver.close()
+                await client.close()
+                await asyncio.wait_for(asyncio.gather(*tasks), 1)
+        events = [json.loads(row.split('relay_room ', 1)[1]) for row in logs.output
+                  if 'relay_room ' in row]
+        streams = {row['role']: row for row in events if row['event'] == 'stream_close'}
+        self.assertEqual(set(streams), {'client', 'receiver'})
+        for role, payload in (('client', client_payload), ('receiver', receiver_payload)):
+            row = streams[role]
+            self.assertEqual(row['ingressFrames'], 1)
+            self.assertEqual(row['ingressBytes'], len(payload))
+            self.assertEqual(row['egressCompletedFrames'], 1)
+            self.assertEqual(row['egressCompletedBytes'], len(payload))
+            self.assertLessEqual(row['firstIngressUtcMs'], row['lastEgressCompletedUtcMs'])
+        self.assertNotIn('SECRET-ROOM', '\n'.join(logs.output))
+        self.assertNotIn('ciphertext-secret', '\n'.join(logs.output))
+
     async def test_stream_lifetime_and_forward_timeout_are_distinct(self):
         for expected in ('stream_lifetime_expiry', 'forward_send_timeout'):
             with self.subTest(reason=expected):
@@ -215,6 +247,13 @@ async def test_stream_lifetime_and_forward_timeout_are_distinct(self):
                         await asyncio.gather(*tasks, return_exceptions=True)
                 self.assertTrue(any('"reason": "'+expected+'"' in row for row in logs.output))
                 self.assertTrue(any('"closeCode": 1000' in row for row in logs.output))
+                if expected == 'forward_send_timeout':
+                    streams = [json.loads(row.split('relay_room ', 1)[1]) for row in logs.output
+                               if '"event": "stream_close"' in row and '"role": "client"' in row]
+                    self.assertEqual(len(streams), 1)
+                    self.assertEqual(streams[0]['ingressFrames'], 1)
+                    self.assertEqual(streams[0]['egressCompletedFrames'], 0)
+                    self.assertIsNone(streams[0]['firstEgressCompletedUtcMs'])
                 self.assertNotIn('SECRET-ROOM', '\n'.join(logs.output))