fix(channels): serialize Feishu connect completion

This commit is contained in:
chengyongru
2026-07-27 01:00:12 +08:00
committed by Xubin Ren
parent 4835814746
commit 2e2f15dd0c
2 changed files with 81 additions and 20 deletions
+24 -20
View File
@@ -5,6 +5,7 @@ from __future__ import annotations
import asyncio import asyncio
import json import json
import secrets import secrets
import threading
import time import time
from dataclasses import dataclass from dataclasses import dataclass
from typing import Any from typing import Any
@@ -41,6 +42,7 @@ class FeishuConnectStore:
def __init__(self) -> None: def __init__(self) -> None:
self._sessions: dict[str, FeishuConnectSession] = {} self._sessions: dict[str, FeishuConnectSession] = {}
self._completion_lock = threading.Lock()
async def handle(self, action: str, query: QueryParams) -> dict[str, Any]: async def handle(self, action: str, query: QueryParams) -> dict[str, Any]:
"""Handle one generic settings connection action.""" """Handle one generic settings connection action."""
@@ -58,7 +60,7 @@ class FeishuConnectStore:
if action == "poll": if action == "poll":
return await asyncio.to_thread(self.poll, session_id) return await asyncio.to_thread(self.poll, session_id)
if action == "cancel": if action == "cancel":
return self.cancel(session_id) return await asyncio.to_thread(self.cancel, session_id)
raise ChannelConnectError(f"unsupported Feishu connect action: {action}", status=404) raise ChannelConnectError(f"unsupported Feishu connect action: {action}", status=404)
def start( def start(
@@ -129,28 +131,29 @@ class FeishuConnectStore:
status = result.get("status") status = result.get("status")
if status == "succeeded": if status == "succeeded":
if self._sessions.get(session_id) is not session: with self._completion_lock:
if self._sessions.get(session_id) is not session:
return {
"session_id": session_id,
"instance_id": session.instance_id,
"status": "cancelled",
"message": "Feishu connection cancelled.",
}
session.domain = str(result.get("domain") or session.domain)
session.instance_id = feishu.save_registration_result(
result,
instance_id=session.instance_id,
name=session.instance_name,
)
self._sessions.pop(session_id, None)
return { return {
"session_id": session_id, "session_id": session_id,
"instance_id": session.instance_id, "instance_id": session.instance_id,
"status": "cancelled", "status": "succeeded",
"message": "Feishu connection cancelled.", "message": "Feishu is connected.",
"domain": session.domain,
"app_id": result.get("app_id"),
} }
session.domain = str(result.get("domain") or session.domain)
session.instance_id = feishu.save_registration_result(
result,
instance_id=session.instance_id,
name=session.instance_name,
)
self._sessions.pop(session_id, None)
return {
"session_id": session_id,
"instance_id": session.instance_id,
"status": "succeeded",
"message": "Feishu is connected.",
"domain": session.domain,
"app_id": result.get("app_id"),
}
session.domain = str(result.get("domain") or session.domain) session.domain = str(result.get("domain") or session.domain)
if status == "failed": if status == "failed":
@@ -166,7 +169,8 @@ class FeishuConnectStore:
return _pending_payload(session) return _pending_payload(session)
def cancel(self, session_id: str) -> dict[str, Any]: def cancel(self, session_id: str) -> dict[str, Any]:
session = self._sessions.pop(session_id, None) with self._completion_lock:
session = self._sessions.pop(session_id, None)
return { return {
"session_id": session_id, "session_id": session_id,
"instance_id": session.instance_id if session else DEFAULT_INSTANCE_ID, "instance_id": session.instance_id if session else DEFAULT_INSTANCE_ID,
@@ -63,3 +63,60 @@ async def test_feishu_cancel_wins_over_inflight_confirmation(
assert cancelled["status"] == "cancelled" assert cancelled["status"] == "cancelled"
assert completed["status"] == "cancelled" assert completed["status"] == "cancelled"
assert saved_results == [] assert saved_results == []
@pytest.mark.asyncio
async def test_feishu_cancel_does_not_interleave_with_registration_save(
monkeypatch: pytest.MonkeyPatch,
) -> None:
save_started = threading.Event()
release_save = threading.Event()
monkeypatch.setattr(feishu, "_init_registration", lambda _domain: None)
monkeypatch.setattr(
feishu,
"_begin_registration",
lambda _domain: {
"device_code": "device-lock",
"qr_url": "https://qr.example/lock",
"expire_in": 600,
"interval": 2,
},
)
monkeypatch.setattr(
feishu,
"poll_registration_once",
lambda **_kwargs: {
"status": "succeeded",
"domain": "feishu",
"app_id": "saved-app",
"app_secret": "saved-secret",
},
)
def fake_save_registration_result(
_result: dict[str, Any],
**_kwargs: Any,
) -> str:
save_started.set()
assert release_save.wait(timeout=5)
return "default"
monkeypatch.setattr(feishu, "save_registration_result", fake_save_registration_result)
store = FeishuConnectStore()
started = await store.handle("start", {})
query = {"session_id": [started["session_id"]]}
poll_task = asyncio.create_task(store.handle("poll", query))
assert await asyncio.to_thread(save_started.wait, 5)
cancel_task = asyncio.create_task(store.handle("cancel", query))
await asyncio.sleep(0)
assert not cancel_task.done()
release_save.set()
completed = await poll_task
cancelled = await cancel_task
assert completed["status"] == "succeeded"
assert cancelled["status"] == "cancelled"