Nexus/docs/design/plans/2026-05-17-retry-progress.md

推送失败重试 + 实时进度 实现计划

⚠️ 已归档 — 2026-05-22 本文档引用的文件结构已过时：server/database.py 已拆分为 server/domain/models/；PHP 前端 (retry.php/servers.php) 已迁移到 Tailwind CSS v4 + Alpine.js (web/app/retries.html/servers.html)；datetime.utcnow() 已弃用，代码已改为 datetime.now(timezone.utc)。功能已实现，保留本文档供历史参考。

面向 AI 代理的工作者： 使用 superpowers:executing-plans 逐任务实现。

目标： 失败推送自动 5 分钟重试 + rsync dry-run 一致性验证后消失 + 推送时 WebSocket 实时进度条

架构： PushRetryJob MySQL 表驱动重试队列；retry_engine 后台 60s 扫描；sync_engine 流式解析 rsync 逐文件回调；WebSocket 推送进度到前端实时进度条

技术栈： Python FastAPI + SQLAlchemy + MySQL + WebSocket + PHP 7.4

---

任务 1：PushRetryJob 模型 + retry_engine 后台循环

文件： server/database.py、server/services/retry_engine.py（新）、server/main.py

• 步骤 1：database.py 添加 PushRetryJob 模型

在 PasswordPreset 类之后添加：

class PushRetryJob(Base):
    """Failed push retry queue — auto-retry every 5 min until consistency"""
    __tablename__ = "push_retry_jobs"

    id = Column(Integer, primary_key=True, autoincrement=True)
    server_id = Column(Integer, ForeignKey("servers.id"), nullable=False)
    server_name = Column(String(100), comment="冗余字段")
    source_path = Column(String(500), nullable=False)
    target_path = Column(String(500), nullable=False)
    status = Column(String(20), default="pending")  # pending/retrying/abandoned
    retry_count = Column(Integer, default=0)
    max_retries = Column(Integer, default=100)
    next_retry_at = Column(DateTime)
    last_error = Column(Text)
    created_at = Column(DateTime, default=datetime.datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.datetime.utcnow)

• 步骤 2：创建 retry_engine.py

"""Retry engine — auto-retry failed pushes every 5 min, verify consistency"""
import asyncio
from datetime import datetime, timedelta
from server.database import SessionLocal, PushRetryJob, Server
from server.services.sync_engine import push_to_server


async def verify_consistency(server, source_path: str, target_path: str) -> bool:
    """Dry-run rsync to check if source and target are identical"""
    import re
    ssh_opts = f"-p {server.port} -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null -o ConnectTimeout=10"
    if server.auth_method == "key" and server.ssh_key_path:
        ssh_opts += f" -i {server.ssh_key_path}"

    cmd = [
        "rsync", "-avzn", "--stats", "--delete",
        "-e", f"ssh {ssh_opts}",
        f"{source_path.rstrip('/')}/",
        f"{server.username}@{server.domain}:{target_path.rstrip('/')}/",
    ]
    proc = await asyncio.create_subprocess_exec(*cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
    stdout, _ = await proc.communicate()
    output = stdout.decode()
    for line in output.split("\n"):
        if m := re.search(r'Number of files:\s*([\d,]+)', line):
            count = int(m.group(1).replace(",", ""))
            if count == 0:
                return True
    return False


async def retry_loop():
    """Background loop: scan retry jobs every 60s"""
    await asyncio.sleep(5)  # 启动后等 5 秒再开始
    while True:
        db = SessionLocal()
        try:
            now = datetime.utcnow()
            jobs = db.query(PushRetryJob).filter(
                PushRetryJob.status == "pending",
                PushRetryJob.next_retry_at <= now,
            ).all()
            for job in jobs:
                job.status = "retrying"
                db.commit()
                server = db.query(Server).filter(Server.id == job.server_id).first()
                if not server:
                    job.status = "abandoned"
                    job.last_error = "Server deleted"
                    db.commit()
                    continue
                try:
                    result = await push_to_server(server, job.source_path, job.target_path)
                    if result.get("status") == "success":
                        consistent = await verify_consistency(server, job.source_path, job.target_path)
                        if consistent:
                            db.delete(job)
                        else:
                            _reschedule(job, result.get("message", "Not consistent"))
                    else:
                        _reschedule(job, result.get("message", "Unknown error"))
                except Exception as e:
                    _reschedule(job, str(e))
                db.commit()
        finally:
            db.close()
        await asyncio.sleep(60)


def _reschedule(job, error: str):
    job.status = "pending"
    job.retry_count += 1
    if job.retry_count >= job.max_retries:
        job.status = "abandoned"
    job.next_retry_at = datetime.utcnow() + timedelta(minutes=5)
    job.last_error = error[:500]

• 步骤 3：main.py 启动 retry_loop

在 lifespan 函数末尾（background_tasks 创建处）添加：

from server.services.retry_engine import retry_loop
background_tasks.append(asyncio.create_task(retry_loop()))

• 步骤 4：验证 + Commit

python -m py_compile server/database.py server/services/retry_engine.py server/main.py && echo "OK"
git add server/database.py server/services/retry_engine.py server/main.py
git commit -m "feat(retry): PushRetryJob 模型 + retry_engine 后台 5 分钟自动重试"

---

任务 2：sync_engine 流式 rsync + pushendpoint 失败时写入 job

文件： server/services/sync_engine.py、server/api/servers.py

• 步骤 1：sync_engine 改 push_to_server — 支持 progress_callback

修改 push_to_server 签名：

async def push_to_server(server, source_path: str, target_path: str, progress_callback=None):

在 proc = await asyncio.create_subprocess_exec(...) 中，改用 stderr=asyncio.subprocess.PIPE 并通过 --out-format='%i %n' 逐行解析：

cmd = [
    "rsync", "-avz", "--stats", "--out-format=%i %n",
    "-e", rsh,
    f"{source_path.rstrip('/')}/",
    f"{server.username}@{server.domain}:{target_path.rstrip('/')}/",
]

proc = await asyncio.create_subprocess_exec(
    *cmd,
    stdout=asyncio.subprocess.PIPE,
    stderr=asyncio.subprocess.PIPE,
)

transferred = 0
while True:
    line = await proc.stdout.readline()
    if not line:
        break
    decoded = line.decode().strip()
    if decoded and not decoded.startswith(("receiving", "sent", "total", "Number", " ")):
        transferred += 1
        if progress_callback:
            try:
                await progress_callback(server.id, {"file": decoded, "transferred": transferred})
            except Exception:
                pass

await proc.wait()
# 解析 --stats 获取总数
stdout_remaining = (await proc.stdout.read()).decode()
# ... parse files_total from stats ...

• 步骤 2：servers.py push 端点 — 失败时自动创建 retry job

在 push_to_servers 函数中，处理失败结果时写 job：

for s, r in zip(servers_list, results):
    if isinstance(r, dict) and r.get("status") != "success":
        existing = db.query(PushRetryJob).filter(
            PushRetryJob.server_id == s.id,
            PushRetryJob.source_path == req.source_path,
            PushRetryJob.status.in_(["pending", "retrying"]),
        ).first()
        if not existing:
            job = PushRetryJob(
                server_id=s.id, server_name=s.name,
                source_path=req.source_path, target_path=s.target_path or "",
                next_retry_at=datetime.utcnow() + timedelta(minutes=5),
                last_error=str(r.get("message", "")),
            )
            db.add(job)
    db.commit()

导入 timedelta 和 PushRetryJob。

• 步骤 3：验证 + Commit

python -m py_compile server/services/sync_engine.py server/api/servers.py && echo "OK"
git add server/services/sync_engine.py server/api/servers.py
git commit -m "feat(retry): sync_engine 流式解析 + push 失败自动创建 retry job"

---

任务 3：WebSocket 实时进度推送

文件： server/api/servers.py、server/main.py

• 步骤 1：servers.py push 端点 — 接入 WebSocket

修改 push 端点，接受 WebSocket 连接或复用 /ws：

使用 asyncio.gather + 每个 task 传 progress_callback，callback 通过 websocket.send_json() 推送进度。

在 servers.py 顶部添加 WS manager：

# WebSocket 进度管理器
_progress_clients: dict = {}  # {task_group_id: [websocket, ...]}

async def broadcast_progress(server_id: int, data: dict):
    for ws_list in _progress_clients.values():
        for ws in ws_list:
            try:
                await ws.send_json({"server_id": server_id, **data})
            except Exception:
                pass

push 端点接受可选 WebSocket 参数，推送 {server_id, file, transferred, total} JSON。

• 步骤 2：验证 + Commit

python -m py_compile server/api/servers.py server/main.py && echo "OK"
git add server/api/servers.py server/main.py
git commit -m "feat(retry): WebSocket 实时进度推送"

---

任务 4：retry.php 重试队列页面 + API 端点

文件： web/retry.php（新）、server/api/servers.py

• 步骤 1：servers.py 加重试队列 API 端点

@router.get("/retry-jobs")
def list_retry_jobs(status: Optional[str] = None, db: Session = Depends(get_db)):
    q = db.query(PushRetryJob).order_by(PushRetryJob.created_at.desc())
    if status:
        q = q.filter(PushRetryJob.status == status)
    return q.all()

@router.post("/retry-jobs/{job_id}/retry-now")
async def retry_now(job_id: int, db: Session = Depends(get_db)):
    job = db.query(PushRetryJob).filter(PushRetryJob.id == job_id).first()
    if not job:
        raise HTTPException(status_code=404)
    server = db.query(Server).filter(Server.id == job.server_id).first()
    if not server:
        raise HTTPException(status_code=404)
    result = await push_to_server(server, job.source_path, job.target_path)
    if result.get("status") == "success":
        db.delete(job)
    else:
        job.retry_count += 1
        job.next_retry_at = datetime.utcnow() + timedelta(minutes=5)
    db.commit()
    return result

@router.delete("/retry-jobs/{job_id}")
def abandon_job(job_id: int, db: Session = Depends(get_db)):
    job = db.query(PushRetryJob).filter(PushRetryJob.id == job_id).first()
    if not job:
        raise HTTPException(status_code=404)
    job.status = "abandoned"
    db.commit()
    return {"message": "abandoned"}

@router.post("/retry-jobs/retry-all")
async def retry_all(db: Session = Depends(get_db)):
    jobs = db.query(PushRetryJob).filter(PushRetryJob.status == "pending").all()
    results = []
    for job in jobs:
        server = db.query(Server).filter(Server.id == job.server_id).first()
        if server:
            r = await push_to_server(server, job.source_path, job.target_path)
            results.append({"job_id": job.id, "server": job.server_name, **r})
            if r.get("status") == "success":
                db.delete(job)
            else:
                job.retry_count += 1
                job.next_retry_at = datetime.utcnow() + timedelta(minutes=5)
    db.commit()
    return results

• 步骤 2：创建 retry.php

<?php require_once __DIR__ . '/auth.php'; require_once __DIR__ . '/api_client.php';
$jobs = []; try { $resp = api()->request('GET', '/api/servers/retry-jobs'); $jobs = $resp['data'] ?? []; } catch(Exception $e) {}
$pending = count(array_filter($jobs, fn($j) => $j['status'] === 'pending'));
?>
<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8"><meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>重试队列 - Nexus</title>
    <link rel="stylesheet" href="style.css">
    <link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/font-awesome/6.5.1/css/all.min.css">
</head>
<body>
<div class="app">
<nav class="sidebar">
    <div class="sidebar-header"><h1><i class="fas fa-sync-alt"></i> Nexus</h1></div>
    <ul class="nav-menu">
        <li class="nav-section-title">主菜单</li>
        <li><a href="index.php"><i class="fas fa-tachometer-alt"></i> <span>仪表盘</span></a></li>
        <li><a href="servers.php"><i class="fas fa-server"></i> <span>服务器管理</span></a></li>
        <li class="active"><a href="retry.php"><i class="fas fa-redo-alt"></i> <span>重试队列</span><?php if($pending): ?> <span class="badge badge-warning"><?=$pending?></span><?php endif; ?></a></li>
        <li><a href="push.php"><i class="fas fa-history"></i> <span>推送日志</span></a></li>
        <li class="nav-section-title">系统</li>
        <li><a href="deploy.php"><i class="fas fa-rocket"></i> <span>部署 Agent</span></a></li>
        <li><a href="settings.php"><i class="fas fa-cog"></i> <span>系统设置</span></a></li>
        <li><a href="logout.php"><i class="fas fa-sign-out-alt"></i> <span>退出登录</span></a></li>
    </ul>
</nav>
<main class="main-content">
<header class="top-bar"><h2><i class="fas fa-redo-alt"></i> 重试队列 (<?=$pending?>)</h2></header>

<div class="quick-actions" style="margin-bottom:16px;">
    <button class="btn btn-warning" onclick="retryAll()"><i class="fas fa-play"></i> 全部重试</button>
</div>

<div class="table-responsive">
<table class="data-table">
<thead><tr><th>服务器</th><th>源目录</th><th>目标目录</th><th>状态</th><th>重试</th><th>下次重试</th><th>操作</th></tr></thead>
<tbody>
<?php if(empty($jobs)): ?>
<tr><td colspan="7" class="empty">无重试任务 🎉</td></tr>
<?php else: foreach($jobs as $j): ?>
<tr>
    <td><?=htmlspecialchars($j['server_name'])?></td>
    <td class="path-cell"><?=htmlspecialchars($j['source_path'])?></td>
    <td class="path-cell"><?=htmlspecialchars($j['target_path'])?></td>
    <td><span class="badge <?=$j['status']==='abandoned'?'badge-danger':($j['status']==='retrying'?'badge-warning':'badge-info')?>"><?=['pending'=>'等待','retrying'=>'重试中','abandoned'=>'已放弃'][$j['status']]??$j['status']?></span></td>
    <td><?=$j['retry_count']?>/<?=$j['max_retries']?></td>
    <td><?=$j['next_retry_at']?></td>
    <td>
        <?php if($j['status'] !== 'abandoned'): ?>
        <button class="btn btn-sm btn-success" onclick="retryNow(<?=$j['id']?>)"><i class="fas fa-play"></i></button>
        <?php endif; ?>
        <button class="btn btn-sm btn-danger" onclick="abandonJob(<?=$j['id']?>)"><i class="fas fa-times"></i></button>
    </td>
</tr>
<?php endforeach; endif; ?>
</tbody></table></div>
</main></div>
<script>
const API_BASE = <?=json_encode(defined('API_BASE_URL')?API_BASE_URL:'http://localhost:8600')?>;
const API_KEY = <?=json_encode(defined('API_KEY')?API_KEY:'')?>;

function retryNow(id){fetch(API_BASE+'/api/servers/retry-jobs/'+id+'/retry-now',{method:'POST',headers:{'X-API-Key':API_KEY}}).then(()=>location.reload());}
function retryAll(){fetch(API_BASE+'/api/servers/retry-jobs/retry-all',{method:'POST',headers:{'X-API-Key':API_KEY}}).then(()=>location.reload());}
function abandonJob(id){if(!confirm('放弃此重试任务？'))return;fetch(API_BASE+'/api/servers/retry-jobs/'+id,{method:'DELETE',headers:{'X-API-Key':API_KEY}}).then(()=>location.reload());}
</script>
</body></html>

• 步骤 3：验证 + Commit

python -m py_compile server/api/servers.py && echo "OK"
git add server/api/servers.py web/retry.php
git commit -m "feat(retry): retry.php 重试队列页面 + 5 个 API 端点"

---

任务 5：servers.php 推送对话框实时进度条

文件： web/servers.php

• 步骤 1：推送对话框改为 WebSocket 实时进度

修改 doPush() 函数，连接 /ws 后收到推送进度更新：

function doPush() {
    // ... 现有逻辑获取 ids 和 sp ...

    document.getElementById('pushOverlay').classList.remove('active');
    // 显示进度弹窗
    showProgressModal(ids);

    const wsProto = API_BASE.startsWith('https') ? 'wss' : 'ws';
    const wsUrl = wsProto + '://' + API_BASE.replace(/^https?:\/\//, '') + '/ws';
    const ws = new WebSocket(wsUrl);

    ws.onopen = () => {
        ws.send(JSON.stringify({ type: 'push', source_path: sp, server_ids: ids }));
    };

    ws.onmessage = (e) => {
        const data = JSON.parse(e.data);
        if (data.type === 'progress') {
            updateProgress(data.server_id, data);
        } else if (data.type === 'done') {
            ws.close();
            setTimeout(() => { closeProgressModal(); location.reload(); }, 1000);
        }
    };

    ws.onerror = () => { closeProgressModal(); alert('推送连接失败'); };
}

进度弹窗 HTML（简化版，放在页面中）：

<div class="push-overlay" id="progressOverlay">
    <div class="push-dialog">
        <h3>推送中...</h3>
        <div class="push-body" id="progressBody"></div>
    </div>
</div>

• 步骤 2：验证 + Commit

git add web/servers.php
git commit -m "feat(retry): 推送对话框 WebSocket 实时进度条"

---

任务 6：sidebar 角标 + 先清理死代码

文件： 全部页面 sidebar、web/db.php、web/install.php、agent/agent.py、agent/agent.sh

• 步骤 1：所有页面 sidebar 加重试队列链接

在 push.php、index.php、settings.php、deploy.php、logs.php 的 sidebar 中，在推送日志前插入：

<li><a href="retry.php"><i class="fas fa-redo-alt"></i> <span>重试队列</span></a></li>

• 步骤 2：清理审计中发现的死代码

删除 web/db.php 中所有 SyncTask 方法。删除 web/install.php 中 sync_tasks 建表。删除 agent/agent.py 中 TriggerFileHandler + start_watchers。

• 步骤 3：全量编译验证 + Commit

find server -name "*.py" | xargs python -m py_compile && echo "All OK"
git add -A && git commit -m "feat(retry): sidebar 角标 + 清理死代码"

---

自检

• 规格全覆盖：PushRetryJob ✓ retry_engine ✓ 一致性验证 ✓ WebSocket 进度 ✓ retry.php ✓ sidebar ✓
• 无占位符/TODO
• 类型一致：push_to_server 签名在 retry_engine 和 servers.py 中一致
• retry API 端点命名一致