fix: settle quota and bound compression work

src/worker/mod.rs

@@ -28,7 +28,7 @@ pub async fn run(state: AppState) -> Result<(), AppError> {
     crate::services::bootstrap::ensure_schema(&state).await?;
 
     let consumer = format!("worker_{}", Uuid::new_v4());
-    ensure_group(&state, &consumer).await?;
+    ensure_group(&state).await?;
 
     let mut last_maintenance = Instant::now();
 
@@ -47,7 +47,7 @@ pub async fn run(state: AppState) -> Result<(), AppError> {
     }
 }
 
-async fn ensure_group(state: &AppState, _consumer: &str) -> Result<(), AppError> {
+async fn ensure_group(state: &AppState) -> Result<(), AppError> {
     let mut conn = state.redis.clone();
 
     let res: Result<redis::Value, redis::RedisError> = redis::cmd("XGROUP")
@@ -284,6 +284,7 @@ async fn mark_task_dead_letter(
     tx.commit()
         .await
         .map_err(|err| AppError::new(ErrorCode::Internal, "提交死信事务失败").with_source(err))?;
+    quota::settle_anonymous_task_reservation(state, task_id).await?;
     Ok(())
 }
 
@@ -384,7 +385,12 @@ async fn process_task(state: &AppState, task_id: Uuid) -> Result<(), AppError> {
         return Ok(());
     };
 
-    if matches!(task.status.as_str(), "completed" | "failed" | "cancelled") {
+    if task.status == "cancelled" {
+        finalize_task_status(state, task_id).await?;
+        return Ok(());
+    }
+    if matches!(task.status.as_str(), "completed" | "failed") {
+        quota::settle_anonymous_task_reservation(state, task_id).await?;
         return Ok(());
     }
 
@@ -752,6 +758,18 @@ async fn finalize_file(
         .await
         .map_err(|err| AppError::new(ErrorCode::Internal, "开启事务失败").with_source(err))?;
 
+    let task_status: Option<String> =
+        sqlx::query_scalar("SELECT status::text FROM tasks WHERE id = $1 FOR UPDATE")
+            .bind(task_id)
+            .fetch_optional(&mut *tx)
+            .await
+            .map_err(|err| {
+                AppError::new(ErrorCode::Internal, "锁定任务状态失败").with_source(err)
+            })?;
+    if !matches!(task_status.as_deref(), Some("pending" | "processing")) {
+        return Err(AppError::new(ErrorCode::InvalidRequest, "任务已结束"));
+    }
+
     // Paid users: charge before marking file completed (atomic w/ status update).
     if charge_units {
         if let Some(billing) = billing_ctx {
@@ -923,11 +941,18 @@ async fn finalize_task_status(state: &AppState, task_id: Uuid) -> Result<(), App
         .await;
 
         let _ = sqlx::query(
-            "UPDATE tasks SET failed_files = GREATEST(total_files - completed_files, 0), completed_at = NOW() WHERE id = $1 AND completed_at IS NULL",
+            r#"
+            UPDATE tasks
+            SET completed_files = (SELECT COUNT(*) FROM task_files WHERE task_id = $1 AND status = 'completed'),
+                failed_files = (SELECT COUNT(*) FROM task_files WHERE task_id = $1 AND status = 'failed'),
+                completed_at = COALESCE(completed_at, NOW())
+            WHERE id = $1
+            "#,
         )
         .bind(task_id)
         .execute(&state.db)
         .await;
+        quota::settle_anonymous_task_reservation(state, task_id).await?;
         return Ok(());
     }
 
@@ -945,6 +970,7 @@ async fn finalize_task_status(state: &AppState, task_id: Uuid) -> Result<(), App
         .execute(&state.db)
         .await
         .map_err(|err| AppError::new(ErrorCode::Internal, "更新任务状态失败").with_source(err))?;
+        quota::settle_anonymous_task_reservation(state, task_id).await?;
     }
 
     Ok(())
@@ -995,12 +1021,34 @@ async fn charge_one_unit(
 }
 
 async fn maintenance(state: &AppState) -> Result<(), AppError> {
+    settle_finished_anonymous_reservations(state).await?;
     cleanup_expired_tasks(state).await?;
     cleanup_stale_zip_temp(state).await?;
     cleanup_expired_records(state).await?;
     Ok(())
 }
 
+async fn settle_finished_anonymous_reservations(state: &AppState) -> Result<(), AppError> {
+    let task_ids: Vec<Uuid> = sqlx::query_scalar(
+        r#"
+        SELECT id
+        FROM tasks
+        WHERE anonymous_units_reserved > 0
+          AND status IN ('completed', 'failed', 'cancelled')
+        ORDER BY completed_at ASC NULLS FIRST
+        LIMIT 200
+        "#,
+    )
+    .fetch_all(&state.db)
+    .await
+    .map_err(|err| AppError::new(ErrorCode::Internal, "查询待结算匿名任务失败").with_source(err))?;
+
+    for task_id in task_ids {
+        quota::settle_anonymous_task_reservation(state, task_id).await?;
+    }
+    Ok(())
+}
+
 async fn cleanup_stale_zip_temp(state: &AppState) -> Result<(), AppError> {
     let root = std::path::Path::new(&state.config.storage_path).join("tmp/zips");
     let mut entries = match tokio::fs::read_dir(&root).await {
@@ -1092,6 +1140,15 @@ async fn cleanup_expired_tasks(state: &AppState) -> Result<(), AppError> {
 }
 
 async fn cleanup_expired_task(state: &AppState, task_id: Uuid) -> Result<(), AppError> {
+    sqlx::query(
+        "UPDATE tasks SET status = 'cancelled', completed_at = COALESCE(completed_at, NOW()) WHERE id = $1 AND expires_at < NOW() AND status IN ('pending', 'processing')",
+    )
+    .bind(task_id)
+    .execute(&state.db)
+    .await
+    .map_err(|err| AppError::new(ErrorCode::Internal, "终止过期任务失败").with_source(err))?;
+    quota::settle_anonymous_task_reservation(state, task_id).await?;
+
     let files: Vec<CleanupFileRow> = sqlx::query_as(
         r#"
         SELECT storage_backend, storage_endpoint_id,