Nexus Agent
5ed3c28491
rsync 抛错时 sync_log 不再永久 running；reconcile 兼容 naive UTC。 Co-authored-by: Cursor <cursoragent@cursor.com>
127 lines
4.5 KiB
Python
127 lines
4.5 KiB
Python
"""Nexus — SyncLog Repository (Async SQLAlchemy implementation)"""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Optional, List, Tuple
|
|
from sqlalchemy import select, func
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from server.domain.models import SyncLog, Server
|
|
|
|
_STALE_RUNNING_MESSAGE = (
|
|
"推送状态异常：长时间未完成（已自动订正，可能因历史缺陷未落库终态）"
|
|
)
|
|
|
|
|
|
def _utc_aware(dt: datetime) -> datetime:
|
|
"""MySQL DateTime columns are naive UTC; normalize for timedelta math."""
|
|
if dt.tzinfo is None:
|
|
return dt.replace(tzinfo=timezone.utc)
|
|
return dt.astimezone(timezone.utc)
|
|
|
|
|
|
class SyncLogRepositoryImpl:
|
|
"""Async SQLAlchemy implementation of SyncLog data access"""
|
|
|
|
def __init__(self, session: AsyncSession):
|
|
self.session = session
|
|
|
|
async def get_by_server_id(self, server_id: int, limit: int = 50) -> List[SyncLog]:
|
|
result = await self.session.execute(
|
|
select(SyncLog)
|
|
.where(SyncLog.server_id == server_id)
|
|
.order_by(SyncLog.started_at.desc())
|
|
.limit(limit)
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
async def get_all_paginated(
|
|
self,
|
|
server_id: Optional[int] = None,
|
|
status: Optional[str] = None,
|
|
sync_mode: Optional[str] = None,
|
|
trigger_type: Optional[str] = None,
|
|
offset: int = 0,
|
|
limit: int = 50,
|
|
) -> Tuple[list, int]:
|
|
"""Global sync log list with filters + server name join + total count."""
|
|
filters = []
|
|
if server_id:
|
|
filters.append(SyncLog.server_id == server_id)
|
|
if status:
|
|
filters.append(SyncLog.status == status)
|
|
if sync_mode:
|
|
filters.append(SyncLog.sync_mode == sync_mode)
|
|
if trigger_type:
|
|
filters.append(SyncLog.trigger_type == trigger_type)
|
|
|
|
# Count
|
|
count_stmt = select(func.count(SyncLog.id))
|
|
if filters:
|
|
count_stmt = count_stmt.where(*filters)
|
|
total = int((await self.session.execute(count_stmt)).scalar() or 0)
|
|
|
|
# Data (with server name join)
|
|
data_stmt = (
|
|
select(SyncLog, Server.name.label("server_name"))
|
|
.join(Server, SyncLog.server_id == Server.id, isouter=True)
|
|
.order_by(SyncLog.started_at.desc())
|
|
.offset(offset)
|
|
.limit(limit)
|
|
)
|
|
if filters:
|
|
data_stmt = data_stmt.where(*filters)
|
|
rows = (await self.session.execute(data_stmt)).all()
|
|
return rows, total
|
|
|
|
async def create(self, log: SyncLog) -> SyncLog:
|
|
self.session.add(log)
|
|
await self.session.commit()
|
|
return log
|
|
|
|
async def update(self, log: SyncLog) -> SyncLog:
|
|
"""Persist in-memory mutations on an existing SyncLog (used after rsync completes)."""
|
|
self.session.add(log)
|
|
await self.session.commit()
|
|
await self.session.refresh(log)
|
|
return log
|
|
|
|
async def reconcile_stale_running(self, max_age_minutes: int = 120) -> int:
|
|
"""Mark ``running`` logs older than *max_age_minutes* as ``failed``."""
|
|
cutoff = datetime.now(timezone.utc) - timedelta(minutes=max_age_minutes)
|
|
result = await self.session.execute(
|
|
select(SyncLog).where(
|
|
SyncLog.status == "running",
|
|
SyncLog.started_at < cutoff,
|
|
)
|
|
)
|
|
stale = list(result.scalars().all())
|
|
if not stale:
|
|
return 0
|
|
|
|
now = datetime.now(timezone.utc)
|
|
for log in stale:
|
|
log.status = "failed"
|
|
if log.finished_at is None:
|
|
log.finished_at = now
|
|
if not log.error_message:
|
|
log.error_message = _STALE_RUNNING_MESSAGE
|
|
if log.duration_seconds == 0 and log.started_at:
|
|
log.duration_seconds = max(
|
|
0, int((now - _utc_aware(log.started_at)).total_seconds())
|
|
)
|
|
|
|
await self.session.commit()
|
|
return len(stale)
|
|
|
|
# Allowed fields for update_status — prevents arbitrary field injection via **kwargs
|
|
SYNC_LOG_UPDATABLE_FIELDS = {"finished_at", "duration_seconds", "error_message", "diff_summary", "status"}
|
|
|
|
async def update_status(self, id: int, status: str, **kwargs) -> Optional[SyncLog]:
|
|
log = await self.session.get(SyncLog, id)
|
|
if log:
|
|
log.status = status
|
|
for key, value in kwargs.items():
|
|
if key in self.SYNC_LOG_UPDATABLE_FIELDS:
|
|
setattr(log, key, value)
|
|
await self.session.commit()
|
|
return log |