From 1634f57a5667c85d54d720faf9e5f96ba18e88ff Mon Sep 17 00:00:00 2001 From: notwin <811718+notwin@users.noreply.github.com> Date: Wed, 29 Jul 2026 14:45:12 +0900 Subject: [PATCH] fix(weixin): recover from empty stale-context responses --- gateway/platforms/weixin.py | 30 +++++++++++++++-------- tests/gateway/test_weixin.py | 46 +++++++++++++++++++++++++++++++++--- 2 files changed, 63 insertions(+), 13 deletions(-) diff --git a/gateway/platforms/weixin.py b/gateway/platforms/weixin.py index bd25319fc8b4a..06bc3b21ff3b1 100644 --- a/gateway/platforms/weixin.py +++ b/gateway/platforms/weixin.py @@ -93,19 +93,24 @@ RETRY_DELAY_SECONDS = 2 BACKOFF_DELAY_SECONDS = 30 SESSION_EXPIRED_ERRCODE = -14 -RATE_LIMIT_ERRCODE = -2 # iLink frequency limit — backoff and retry +RATE_LIMIT_ERRCODE = -2 # iLink overloads -2 for rate limits and stale/invalid context MESSAGE_DEDUP_TTL_SECONDS = 300 def _is_stale_session_ret( ret: "Optional[int]", errcode: "Optional[int]", errmsg: "Optional[str]", ) -> bool: - """True when iLink returns ret=-2 / errcode=-2 with 'unknown error', - which is a stale-session signal (same as errcode=-14) rather than - a genuine rate limit.""" + """Return whether ``-2`` is iLink's stale-context variant. + + Cron pushes to an inactive peer can return ``ret=-2`` with either an empty + error message or ``"unknown error"``. Both recover when the saved + ``context_token`` is removed and the message is retried tokenless. An + explicit frequency/rate-limit message must continue through the normal + backoff path instead. + """ if ret != RATE_LIMIT_ERRCODE and errcode != RATE_LIMIT_ERRCODE: return False - return (errmsg or "").lower() == "unknown error" + return (errmsg or "").strip().lower() in {"", "unknown error"} MEDIA_IMAGE = 1 @@ -314,6 +319,11 @@ def set(self, account_id: str, user_id: str, token: str) -> None: self._cache[self._key(account_id, user_id)] = token self._persist(account_id) + def delete(self, account_id: str, user_id: str) -> None: + """Remove a stale peer token from memory and durable storage.""" + self._cache.pop(self._key(account_id, user_id), None) + self._persist(account_id) + def _persist(self, account_id: str) -> None: prefix = f"{account_id}:" payload = { @@ -1812,12 +1822,12 @@ async def _send_text_chunk_locked( if is_session_expired and not retried_without_token and context_token: retried_without_token = True context_token = None - self._token_store._cache.pop( - self._token_store._key(self._account_id, chat_id), None - ) + self._token_store.delete(self._account_id, chat_id) logger.warning( - "[%s] session expired for %s; retrying without context_token", - self.name, _safe_id(chat_id), + "[%s] stale context_token for %s " + "(ret=%s errcode=%s errmsg=%r); retrying tokenless", + self.name, _safe_id(chat_id), ret, errcode, + resp.get("errmsg"), ) continue # Rate limit (-2) — backoff and retry diff --git a/tests/gateway/test_weixin.py b/tests/gateway/test_weixin.py index e5343d7e021d1..916f5fd38334b 100644 --- a/tests/gateway/test_weixin.py +++ b/tests/gateway/test_weixin.py @@ -318,6 +318,20 @@ def _boom(_src, _dst): assert json.loads(token_path.read_text(encoding="utf-8")) == {"user-a": "old-token"} warning_mock.assert_called_once() + def test_context_token_delete_removes_memory_and_disk_entry(self, tmp_path): + store = ContextTokenStore(str(tmp_path)) + store.set("acct", "user-a", "stale-token") + store.set("acct", "user-b", "fresh-token") + + store.delete("acct", "user-a") + + assert store.get("acct", "user-a") is None + assert store.get("acct", "user-b") == "fresh-token" + token_path = tmp_path / "weixin" / "accounts" / "acct.context-tokens.json" + assert json.loads(token_path.read_text(encoding="utf-8")) == { + "user-b": "fresh-token" + } + def test_save_sync_buf_preserves_existing_file_on_replace_failure(self, tmp_path, monkeypatch): sync_path = tmp_path / "weixin" / "accounts" / "acct.sync.json" sync_path.parent.mkdir(parents=True, exist_ok=True) @@ -446,6 +460,32 @@ async def flaky_send(*args, **kwargs): assert first_try["text"] == retry["text"] assert first_try["client_id"] == retry["client_id"] + @patch("gateway.platforms.weixin._send_message", new_callable=AsyncMock) + def test_empty_errmsg_minus_2_retries_without_stale_context_token( + self, send_message_mock, tmp_path + ): + adapter = self._connected_adapter() + token_store = ContextTokenStore(str(tmp_path)) + token_store.set("test-account", "wxid_test123", "stale-ctx-token") + adapter._token_store = token_store + send_message_mock.side_effect = [ + {"ret": -2, "errcode": None, "errmsg": None}, + {"ret": 0, "errcode": 0}, + ] + + result = asyncio.run(adapter.send("wxid_test123", "scheduled journal")) + + assert result.success is True + assert send_message_mock.await_count == 2 + first_attempt = send_message_mock.await_args_list[0].kwargs + tokenless_retry = send_message_mock.await_args_list[1].kwargs + assert first_attempt["context_token"] == "stale-ctx-token" + assert tokenless_retry["context_token"] is None + assert token_store.get("test-account", "wxid_test123") is None + token_path = tmp_path / "weixin" / "accounts" / "test-account.context-tokens.json" + assert json.loads(token_path.read_text(encoding="utf-8")) == {} + assert adapter._rate_limit_events == [] + @patch("gateway.platforms.weixin.asyncio.sleep", new_callable=AsyncMock) @patch("gateway.platforms.weixin._send_message", new_callable=AsyncMock) def test_repeated_rate_limits_open_circuit_for_followup_sends(self, send_message_mock, sleep_mock): @@ -958,9 +998,9 @@ def test_ret_minus_2_with_freq_limit_is_not_stale(self): # Genuine rate limit — must NOT be treated as stale session. assert weixin._is_stale_session_ret(-2, None, "freq limit") is False - def test_ret_minus_2_with_no_errmsg_is_not_stale(self): - assert weixin._is_stale_session_ret(-2, None, None) is False - assert weixin._is_stale_session_ret(-2, None, "") is False + def test_ret_minus_2_with_no_errmsg_is_stale(self): + assert weixin._is_stale_session_ret(-2, None, None) is True + assert weixin._is_stale_session_ret(-2, None, "") is True def test_errcode_minus_14_is_not_matched_here(self): # -14 is handled by the separate SESSION_EXPIRED_ERRCODE path; the