momentry_studio/docs/DEEP_SCAN_QUEUE_FEATURE_REQUEST.md

Deep Scan 队列功能需求

背景

当前问题

前端临时解决方案：

• 使用 p-limit 控制并发数（maxConcurrent = 2）
• 每个文件启动扫描后等待 2 秒
• 勉强限制同时扫描的文件数量

存在的限制：

1. 不可靠：前端无法准确控制后端资源使用
2. 效率低：固定延迟 2 秒可能不够或浪费时间
3. 状态不同步：前端和后端对"正在扫描"的理解不一致
4. 无优先级：所有文件按顺序处理，无法优先处理重要文件
5. 无错误恢复：如果某个文件处理失败，没有重试机制
6. 无进度跟踪：无法准确知道队列中每个文件的处理进度

---

需求说明

核心需求

1. 后端队列管理

• 后端维护一个全局的扫描队列
• 限制同时处理的文件数量（可配置，默认 2-4）
• 自动排队等待，先进先出（FIFO）

2. 队列状态 API

• /api/v1/queue/status - 获取队列状态

  {
    "active_count": 2,
    "max_concurrent": 2,
    "pending_count": 3,
    "total_processed": 10,
    "queue": [
      {
        "file_uuid": "xxx",
        "file_name": "video1.mp4",
        "status": "processing",
        "started_at": "2026-10-07T10:00:00Z",
        "progress": 45
      },
      {
        "file_uuid": "yyy",
        "file_name": "video2.mp4",
        "status": "pending",
        "queued_at": "2026-10-07T10:01:00Z",
        "position": 1
      }
    ]
  }
  

3. 队列控制 API

• /api/v1/queue/pause - 暂停队列
• /api/v1/queue/resume - 恢复队列
• /api/v1/queue/clear - 清空队列
• /api/v1/queue/prioritize/{file_uuid} - 提升优先级

4. 动态并发控制

• 根据系统资源（CPU、内存、GPU）自动调整并发数
• 或从配置文件/API 设置 max_concurrent

5. 错误处理与重试

• 自动重试失败的任务（可配置重试次数）
• 记录失败原因
• 提供手动重试 API

---

API 设计建议

1. 提交扫描任务

请求：

POST /api/v1/file/{file_uuid}/deep-scan
Content-Type: application/json

{
  "processors": ["face", "asr", "ocr"],
  "priority": "normal",  // low, normal, high
  "retry_on_failure": true,
  "callback_url": "http://..."  // 可选，完成后通知
}

响应：

{
  "success": true,
  "task_id": "task_xxx",
  "status": "queued",
  "position": 3,
  "estimated_start": "2026-10-07T10:05:00Z"
}

---

2. 查询队列状态

请求：

GET /api/v1/queue/status

响应：

{
  "active_count": 2,
  "max_concurrent": 2,
  "pending_count": 3,
  "paused": false,
  "queue": [
    {
      "task_id": "task_xxx",
      "file_uuid": "xxx",
      "file_name": "video1.mp4",
      "status": "processing",
      "started_at": "2026-10-07T10:00:00Z",
      "progress": 45,
      "processors": ["face", "asr"]
    },
    {
      "task_id": "task_yyy",
      "file_uuid": "yyy",
      "file_name": "video2.mp4",
      "status": "pending",
      "queued_at": "2026-10-07T10:01:00Z",
      "position": 1,
      "priority": "normal"
    }
  ],
  "statistics": {
    "total_processed": 10,
    "total_failed": 2,
    "average_time": 120.5
  }
}

---

3. 任务优先级调整

请求：

PUT /api/v1/queue/task/{task_id}/priority
Content-Type: application/json

{
  "priority": "high"
}

响应：

{
  "success": true,
  "position": 1
}

---

4. 取消任务

请求：

DELETE /api/v1/queue/task/{task_id}

响应：

{
  "success": true,
  "message": "Task cancelled"
}

---

数据库设计

queue_tasks 表

CREATE TABLE queue_tasks (
  task_id TEXT PRIMARY KEY,
  file_uuid TEXT NOT NULL,
  file_name TEXT,
  status TEXT DEFAULT 'pending',  -- pending, processing, completed, failed
  priority TEXT DEFAULT 'normal',
  processors JSONB,
  progress INTEGER DEFAULT 0,
  queued_at TIMESTAMP DEFAULT NOW(),
  started_at TIMESTAMP,
  completed_at TIMESTAMP,
  error_message TEXT,
  retry_count INTEGER DEFAULT 0,
  max_retries INTEGER DEFAULT 3,
  created_at TIMESTAMP DEFAULT NOW(),
  updated_at TIMESTAMP DEFAULT NOW()
);

CREATE INDEX idx_queue_status ON queue_tasks(status);
CREATE INDEX idx_queue_priority ON queue_tasks(priority, queued_at);

---

实现建议

后端架构

Queue Manager (Singleton)
├── Task Queue (Priority Queue)
├── Worker Pool (2-4 workers)
├── Resource Monitor
│   ├── CPU Usage
│   ├── Memory Usage
│   └── GPU Usage
└── Scheduler
    ├── Dynamic Concurrency Adjustment
    └── Task Retry Logic

核心组件

1. QueueManager

pub struct QueueManager {
    tasks: Arc<Mutex<VecDeque<QueueTask>>>,
    active_tasks: Arc<Mutex<HashMap<TaskId, JoinHandle>>>,
    max_concurrent: Arc<AtomicU32>,
    worker_semaphore: Arc<Semaphore>,
}

impl QueueManager {
    pub async fn submit_task(&self, task: QueueTask) -> Result<TaskId>;
    pub async fn get_status(&self) -> QueueStatus;
    pub async fn cancel_task(&self, task_id: TaskId) -> Result<()>;
    pub async fn adjust_concurrency(&self, max: u32);
}

2. ResourceMonitor

pub struct ResourceMonitor {
    cpu_threshold: f64,  // 80%
    memory_threshold: f64,  // 80%
    gpu_threshold: f64,  // 90%
}

impl ResourceMonitor {
    pub fn check_resources(&self) -> ResourceStatus;
    pub fn suggest_concurrency(&self) -> u32;
}

---

前端集成

简化后的前端代码

async function batchDeepScan() {
  const filesToProcess = selectedRegisteredFiles.value
  
  // 提交所有任务到后端队列
  for (const f of filesToProcess) {
    await apiCall('deep_scan', { 
      fileUuid: f.file_uuid,
      priority: 'normal'
    })
  }
  
  // 轮询队列状态
  const interval = setInterval(async () => {
    const status = await apiCall('get_queue_status')
    actionMsg.value = `Processing: ${status.active_count} active, ${status.pending_count} queuing...`
    
    if (status.pending_count === 0 && status.active_count === 0) {
      clearInterval(interval)
      actionMsg.value = 'All tasks completed'
    }
  }, 2000)
}

---

预期效果

用户体验改进

当前（前端队列）：

• ❌ 不准确的并发控制
• ❌ 固定延迟效率低
• ❌ 无法查看队列状态
• ❌ 无法调整优先级
• ❌ 无法恢复失败任务

改进后（后端队列）：

• ✅ 准确的并发控制
• ✅ 自动资源管理
• ✅ 实时队列状态查看
• ✅ 灵活的优先级调整
• ✅ 自动重试机制
• ✅ 错误恢复能力

---

配置选项

momentry.toml

[queue]
# 最大并发处理数
max_concurrent = 2

# 是否启用动态调整
dynamic_concurrency = true

# 资源阈值
cpu_threshold = 80.0
memory_threshold = 80.0
gpu_threshold = 90.0

# 重试配置
max_retries = 3
retry_delay = 10  # seconds

# 任务超时
task_timeout = 3600  # seconds

---

里程碑

Phase 1: 基础队列功能

• 队列数据结构和 API
• 基本的并发控制
• 任务状态跟踪

Phase 2: 高级功能

• 优先级队列
• 资源监控
• 动态并发调整

Phase 3: 错误处理

• 自动重试
• 错误日志
• 手动干预 API

---

参考实现

类似项目：

• Celery (Python)
• Bull (Node.js)
• Sidekiq (Ruby)

关键特性：

• 持久化队列
• 优先级支持
• 进度跟踪
• 错误处理

---

联系方式

需求提出者： Momentry Studio Team
日期： 2026-10-07
优先级： High
预计工期： 2-3 周