145 lines
4.9 KiB
Python
145 lines
4.9 KiB
Python
"""client_agent 空闲会话自动归档后台任务。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import time
|
|
|
|
from .archive_lock import SessionArchiveLock
|
|
from .archive_schedule import SessionArchiveSchedule
|
|
|
|
|
|
logger = logging.getLogger("client_agent.idle_archive_worker")
|
|
|
|
|
|
class IdleArchiveWorker:
|
|
"""扫描到期会话并调用统一 MemoryService 完成归档。"""
|
|
|
|
RETRY_BASE_DELAY = 30
|
|
RETRY_MAX_DELAY = 15 * 60
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
redis,
|
|
memory_service,
|
|
interval: int = 10,
|
|
batch_size: int = 100,
|
|
lock_ttl: int = 120,
|
|
clock=time.time,
|
|
):
|
|
self.redis = redis
|
|
self.memory_service = memory_service
|
|
self.interval = interval
|
|
self.batch_size = batch_size
|
|
self.clock = clock
|
|
self.schedule = SessionArchiveSchedule(redis)
|
|
self.lock = SessionArchiveLock(redis, default_ttl=lock_ttl)
|
|
self._stop_event = asyncio.Event()
|
|
|
|
async def archive_due_sessions(self) -> int:
|
|
"""处理一批到期会话，返回成功归档数量。"""
|
|
session_ids = await self.schedule.list_due(
|
|
now=self.clock(), limit=self.batch_size
|
|
)
|
|
archived_count = 0
|
|
for session_id in session_ids:
|
|
token = await self.lock.acquire(session_id)
|
|
if token is None:
|
|
continue
|
|
try:
|
|
if await self._archive_one(session_id):
|
|
archived_count += 1
|
|
finally:
|
|
await self.lock.release(session_id, token)
|
|
return archived_count
|
|
|
|
async def recover_due_index(self) -> int:
|
|
"""应用启动时从会话元数据恢复归档调度索引。"""
|
|
try:
|
|
return await self.schedule.rebuild_due_index()
|
|
except Exception:
|
|
logger.exception("failed to rebuild idle archive due index")
|
|
return 0
|
|
|
|
async def _archive_one(self, session_id: str) -> bool:
|
|
"""归档单个会话；失败时保留消息并推迟下一次扫描。"""
|
|
owner = await self.redis.get(f"session:{session_id}")
|
|
if owner is None:
|
|
await self.schedule.remove(session_id)
|
|
return False
|
|
owner = owner.decode() if isinstance(owner, bytes) else owner
|
|
meta = await self.schedule.get_meta(session_id)
|
|
retry_count = int(meta.get("archive_retry_count", 0) or 0)
|
|
await self.schedule.mark_processing(session_id)
|
|
try:
|
|
warnings = await self.memory_service.close_session(
|
|
customer_id=int(owner),
|
|
session_id=session_id,
|
|
)
|
|
if warnings:
|
|
await self._schedule_retry(
|
|
session_id,
|
|
retry_count=retry_count + 1,
|
|
error=";".join(warnings),
|
|
)
|
|
return False
|
|
await self.redis.delete(f"session:{session_id}")
|
|
await self.schedule.remove(session_id)
|
|
return True
|
|
except Exception as exc:
|
|
logger.exception("idle session archive failed: session_id=%s", session_id)
|
|
await self._schedule_retry(
|
|
session_id,
|
|
retry_count=retry_count + 1,
|
|
error=f"{type(exc).__name__}: {exc}",
|
|
)
|
|
return False
|
|
|
|
async def _schedule_retry(
|
|
self, session_id: str, *, retry_count: int, error: str
|
|
) -> None:
|
|
"""按指数退避登记下一次归档尝试。"""
|
|
delay = min(
|
|
self.RETRY_BASE_DELAY * (2 ** max(0, retry_count - 1)),
|
|
self.RETRY_MAX_DELAY,
|
|
)
|
|
next_retry_at = self.clock() + delay
|
|
await self.schedule.mark_retry(
|
|
session_id,
|
|
retry_count=retry_count,
|
|
error=error,
|
|
next_retry_at=next_retry_at,
|
|
)
|
|
await self.schedule.reschedule(session_id, due_at=next_retry_at)
|
|
|
|
async def retry_now(self, session_id: str) -> None:
|
|
"""提供手工补偿入口，立即把指定会话放回扫描队列。"""
|
|
await self.redis.hset(
|
|
self.schedule.meta_key(session_id),
|
|
mapping={"archive_status": "pending"},
|
|
)
|
|
await self.schedule.reschedule(session_id, due_at=self.clock())
|
|
|
|
async def run(self) -> None:
|
|
"""启动后台循环，异常不会击穿主应用。"""
|
|
while not self._stop_event.is_set():
|
|
try:
|
|
await self.archive_due_sessions()
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception:
|
|
logger.exception("idle archive scan failed")
|
|
try:
|
|
await asyncio.wait_for(self._stop_event.wait(), timeout=self.interval)
|
|
except asyncio.TimeoutError:
|
|
continue
|
|
|
|
async def stop(self) -> None:
|
|
"""请求后台循环停止。"""
|
|
self._stop_event.set()
|
|
|
|
|
|
__all__ = ["IdleArchiveWorker"]
|