fix: bound large WebP lossless probe cost

.gitea/workflows/ci.yml

@@ -33,10 +33,20 @@ jobs:
       IMAGEFORGE_TEST_REDIS_URL: redis://redis:6379/
       JWT_SECRET: imageforge-ci-jwt-secret
       API_KEY_PEPPER: imageforge-ci-api-key-pepper
-      EXPECTED_EXTERNAL_TESTS: '9'
+      EXPECTED_EXTERNAL_TESTS: '13'
     steps:
       - name: Checkout
-        uses: actions/checkout@v4
+        env:
+          GITEA_JOB_TOKEN: ${{ gitea.token }}
+        run: |
+          set -Eeuo pipefail
+          auth="$(printf 'x-access-token:%s' "$GITEA_JOB_TOKEN" | base64 | tr -d '\n')"
+          git init .
+          git remote add origin "${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}.git"
+          git -c http.extraHeader="Authorization: Basic ${auth}" \
+            fetch --no-tags --depth=1 origin "$GITHUB_REF"
+          git checkout --detach "$GITHUB_SHA"
+          test "$(git rev-parse HEAD)" = "$GITHUB_SHA"
 
       - name: Verify migration line endings
         run: |
@@ -53,14 +63,25 @@ jobs:
             raise SystemExit("migrations must use CRLF: " + ", ".join(invalid))
           PY
 
-      - name: Install Rust toolchain
+      - name: Install native build dependencies
-        uses: dtolnay/rust-toolchain@stable
+        run: |
-        with:
+          apt-get update
-          toolchain: '1.92'
+          DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends \
-          components: rustfmt, clippy
+            cmake libdav1d-dev nasm pkg-config
+          pkg-config --atleast-version=1.3.0 dav1d
+          rm -rf /var/lib/apt/lists/*
 
-      - name: Cache Rust build
+      - name: Install Rust toolchain
-        uses: Swatinem/rust-cache@v2
+        run: |
+          set -Eeuo pipefail
+          curl --proto '=https' --tlsv1.2 --fail --silent --show-error \
+            --location --retry 10 --retry-connrefused https://sh.rustup.rs \
+            | sh -s -- -y --default-toolchain none --profile minimal
+          /root/.cargo/bin/rustup toolchain install 1.92 \
+            --profile minimal --component rustfmt --component clippy --no-self-update
+          /root/.cargo/bin/rustup default 1.92
+          echo /root/.cargo/bin >> "$GITHUB_PATH"
+          /root/.cargo/bin/rustc --version --verbose
 
       - name: Check Rust formatting
         run: cargo fmt --all -- --check
@@ -78,14 +99,22 @@ jobs:
         run: cargo install cargo-audit --locked --version 0.22.2
 
       - name: Audit Rust dependencies
-        run: cargo audit
+        run: |
+          set -Eeuo pipefail
+          audit_db="$RUNNER_TEMP/rustsec-advisory-db"
+          mkdir -p "$audit_db"
+          curl --proto '=https' --tlsv1.2 --fail --silent --show-error \
+            --location --retry 10 --retry-all-errors \
+            https://codeload.github.com/RustSec/advisory-db/tar.gz/refs/heads/main \
+            | tar -xz --strip-components=1 -C "$audit_db"
+          test -f "$audit_db/support.toml"
+          test -d "$audit_db/crates"
+          cargo audit --db "$audit_db" --no-fetch
 
       - name: Install Node.js
-        uses: actions/setup-node@v4
+        run: |
-        with:
+          node --version | grep --extended-regexp '^v22\.'
-          node-version: '22'
+          npm --version
-          cache: npm
-          cache-dependency-path: frontend/package-lock.json
 
       - name: Build frontend
         working-directory: frontend

docker/nginx.conf

@@ -40,6 +40,8 @@ http {
 
     location /downloads/ {
       proxy_pass http://imageforge_api;
+      proxy_read_timeout 300s;
+      proxy_send_timeout 300s;
       proxy_set_header Host $host;
       proxy_set_header X-Real-IP $remote_addr;
       proxy_set_header X-Forwarded-For $remote_addr;

docs/api.md

@@ -276,7 +276,7 @@ Idempotency-Key: <key>          # 建议
 | `output_format` | String | 否 | 输出格式：`png/jpeg/webp/avif/gif/bmp/tiff/ico`（默认保持原格式；ICO 自动等比缩至 256x256 边界） |
 | `max_width` | Integer | 否 | 大于 0 的最大宽度（等比缩放） |
 | `max_height` | Integer | 否 | 大于 0 的最大高度（等比缩放） |
-| `target_size_bytes` | Integer | 否 | 不小于 1024 的目标体积（字节），仅 `jpeg/webp/avif` 输出支持；不能与 `compression_rate` 同时指定 |
+| `target_size_bytes` | Integer | 否 | 不小于 1024 的最大输出体积（字节），仅 `jpeg/webp/avif` 输出支持；系统选择上限内的最高保真结果，不用填充凑到固定大小；不能与 `compression_rate` 同时指定 |
 | `preserve_metadata` | Boolean | 否 | 是否保留 EXIF/ICC（默认 `false`）；元数据输出仅支持 `jpeg/png/webp` |
 
 处理约束：

frontend/src/pages/HomePage.vue

@@ -24,6 +24,7 @@ interface UploadItem {
   status: ItemStatus
   result?: CompressResponse
   error?: string
+  targetSizeBytes?: number
 }
 
 const auth = useAuthStore()
@@ -135,6 +136,28 @@ function getTargetSizeBytes(): number | undefined {
   return Math.round(bytes)
 }
 
+function targetOutputFormat(file: File): OutputFormat {
+  const mime = file.type.trim().toLowerCase()
+  if (mime === 'image/jpeg' || mime === 'image/jpg') return 'jpeg'
+  if (mime === 'image/webp') return 'webp'
+  if (mime === 'image/avif') return 'avif'
+
+  const extension = file.name.split('.').pop()?.toLowerCase()
+  if (extension === 'jpg' || extension === 'jpeg') return 'jpeg'
+  if (extension === 'webp') return 'webp'
+  if (extension === 'avif') return 'avif'
+  return 'webp'
+}
+
+function targetResultHint(item: UploadItem): string | null {
+  if (!item.result || !item.targetSizeBytes) return null
+  const target = formatBytes(item.targetSizeBytes)
+  if (item.result.compressed_size * 4 < item.targetSizeBytes * 3) {
+    return `体积上限 ${target}；当前格式的最高保真结果本身更小，不会添加无效填充。`
+  }
+  return `体积上限 ${target}；结果已控制在上限内。`
+}
+
 function setCompressionMode(mode: CompressionMode) {
   options.mode = mode
   if (
@@ -159,7 +182,7 @@ async function runOne(item: UploadItem) {
     }
 
     const outputFormat: OutputFormat | undefined = options.outputFormat === 'auto'
-      ? (options.mode === 'size' ? 'webp' : undefined)
+      ? (options.mode === 'size' ? targetOutputFormat(item.file) : undefined)
       : options.outputFormat
 
     if (options.mode === 'size' && outputFormat && !targetSizeFormats.has(outputFormat)) {
@@ -188,6 +211,7 @@ async function runOne(item: UploadItem) {
       auth.token,
     )
 
+    item.targetSizeBytes = targetSizeBytes
     item.result = result
     item.status = 'done'
   } catch (err) {
@@ -452,6 +476,9 @@ async function resendVerification() {
                     {{ item.result.saved_percent.toFixed(2) }}%
                   </template>
                 </div>
+                <div v-if="targetResultHint(item)" class="mt-1 text-xs text-slate-500">
+                  {{ targetResultHint(item) }}
+                </div>
                 <div v-if="item.error" class="mt-1 text-xs text-rose-700">{{ item.error }}</div>
               </div>
 
@@ -547,7 +574,7 @@ async function resendVerification() {
                   :class="options.mode === 'size' ? 'bg-indigo-600 text-white' : 'text-slate-600 hover:bg-slate-100'"
                   @click="setCompressionMode('size')"
                 >
-                  按目标大小
+                  按体积上限
                 </button>
               </div>
             </div>
@@ -573,7 +600,7 @@ async function resendVerification() {
 
             <!-- 目标大小模式 -->
             <div v-else class="space-y-1">
-              <div class="text-xs font-medium text-slate-600">目标大小</div>
+              <div class="text-xs font-medium text-slate-600">最大输出大小</div>
               <div class="flex gap-2">
                 <input
                   v-model="options.targetSize"
@@ -591,7 +618,7 @@ async function resendVerification() {
                 </select>
               </div>
               <div class="text-xs text-slate-500">
-                仅支持 JPEG/WebP/AVIF；保持原格式时会自动输出 WebP，过小且无法保证清晰度的目标会被拒绝。
+                这是体积上限，不是固定输出大小。系统会优先使用原格式和最高可用画质；最高画质结果更小时不会填充无效数据。
               </div>
             </div>
 
@@ -601,7 +628,7 @@ async function resendVerification() {
                 v-model="options.outputFormat"
                 class="w-full rounded-md border border-slate-200 bg-white px-3 py-2 text-sm text-slate-800"
               >
-                <option value="auto">{{ options.mode === 'size' ? '自动选择 WebP（推荐）' : '保持原格式（推荐）' }}</option>
+                <option value="auto">{{ options.mode === 'size' ? '智能选择（优先原格式）' : '保持原格式（推荐）' }}</option>
                 <option value="jpeg">JPEG</option>
                 <option value="png" :disabled="options.mode === 'size'">PNG</option>
                 <option value="webp">WebP</option>
@@ -611,7 +638,7 @@ async function resendVerification() {
                 <option value="tiff" :disabled="options.mode === 'size'">TIFF</option>
                 <option value="ico" :disabled="options.mode === 'size'">ICO</option>
               </select>
-              <div class="text-xs text-slate-500">支持按需转码。目标大小模式建议配合 JPEG/WebP/AVIF。</div>
+              <div class="text-xs text-slate-500">支持按需转码。体积上限模式仅使用 JPEG/WebP/AVIF；其他输入会自动转为 WebP。</div>
             </label>
 
             <div class="grid grid-cols-2 gap-3">

frontend/src/pages/admin/AdminBillingPage.vue

@@ -212,7 +212,7 @@ onMounted(async () => {
         >
           {{ subBusy ? '提交中…' : '立即开通' }}
         </button>
-        <span class="text-xs text-slate-500">会取消该用户当前有效订阅，并按月数顺延。</span>
+        <span class="text-xs text-slate-500">会替换当前本地套餐；存在未取消 Stripe 订阅时将拒绝操作。</span>
       </div>
 
       <div v-if="subMessage" class="mt-3 rounded-lg border border-emerald-200 bg-emerald-50 p-3 text-sm text-emerald-900">

migrations/023_cross_provider_subscription_invariant.sql

@@ -0,0 +1,65 @@
+WITH ranked AS (
+  SELECT
+    id,
+    user_id,
+    provider,
+    status::text AS previous_status,
+    current_period_end,
+    ROW_NUMBER() OVER (
+      PARTITION BY user_id
+      ORDER BY
+        CASE WHEN provider = 'stripe' THEN 0 ELSE 1 END,
+        current_period_end DESC,
+        updated_at DESC,
+        id DESC
+    ) AS position
+  FROM subscriptions
+  WHERE status IN ('active', 'trialing', 'past_due')
+), duplicates AS (
+  SELECT *
+  FROM ranked
+  WHERE position > 1
+)
+INSERT INTO audit_logs (
+  user_id, action, resource_type, resource_id, details
+)
+SELECT
+  user_id,
+  'migration_subscription_dedup',
+  'subscription',
+  id,
+  jsonb_build_object(
+    'migration', '023_cross_provider_subscription_invariant',
+    'provider', provider,
+    'previous_status', previous_status,
+    'current_period_end', current_period_end,
+    'reason', 'cross_provider_single_effective_subscription'
+  )
+FROM duplicates;
+
+WITH ranked AS (
+  SELECT
+    id,
+    ROW_NUMBER() OVER (
+      PARTITION BY user_id
+      ORDER BY
+        CASE WHEN provider = 'stripe' THEN 0 ELSE 1 END,
+        current_period_end DESC,
+        updated_at DESC,
+        id DESC
+    ) AS position
+  FROM subscriptions
+  WHERE status IN ('active', 'trialing', 'past_due')
+)
+UPDATE subscriptions AS subscription
+SET status = 'canceled',
+    cancel_at_period_end = false,
+    canceled_at = COALESCE(subscription.canceled_at, NOW()),
+    updated_at = NOW()
+FROM ranked
+WHERE ranked.position > 1
+  AND subscription.id = ranked.id;
+
+CREATE UNIQUE INDEX idx_subscriptions_user_effective_unique
+  ON subscriptions(user_id)
+  WHERE status IN ('active', 'trialing', 'past_due');

migrations/024_task_queue_outbox.sql

@@ -0,0 +1,28 @@
+CREATE TABLE task_queue_outbox (
+  task_id UUID PRIMARY KEY REFERENCES tasks(id) ON DELETE CASCADE,
+  status VARCHAR(20) NOT NULL DEFAULT 'pending',
+  attempts INTEGER NOT NULL DEFAULT 0,
+  next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  lease_owner UUID,
+  lease_until TIMESTAMPTZ,
+  last_error TEXT,
+  delivered_at TIMESTAMPTZ,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  CONSTRAINT task_queue_outbox_status_check
+    CHECK (status IN ('pending', 'delivering', 'delivered', 'dead')),
+  CONSTRAINT task_queue_outbox_attempts_check
+    CHECK (attempts >= 0),
+  CONSTRAINT task_queue_outbox_lease_pair_check
+    CHECK ((lease_owner IS NULL) = (lease_until IS NULL))
+);
+
+CREATE INDEX task_queue_outbox_ready
+  ON task_queue_outbox(next_attempt_at, created_at)
+  WHERE status IN ('pending', 'delivering');
+
+INSERT INTO task_queue_outbox (task_id)
+SELECT id
+FROM tasks
+WHERE status = 'pending'
+ON CONFLICT (task_id) DO NOTHING;

migrations/025_storage_object_lifecycle.sql

@@ -0,0 +1,101 @@
+ALTER TABLE tasks
+  ADD COLUMN deletion_started_at TIMESTAMPTZ,
+  ADD COLUMN deletion_reason VARCHAR(32);
+
+CREATE INDEX tasks_deletion_pending
+  ON tasks(deletion_started_at)
+  WHERE deletion_started_at IS NOT NULL;
+
+CREATE TABLE storage_objects (
+  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+  task_id UUID NOT NULL,
+  task_file_id UUID,
+  object_kind VARCHAR(32) NOT NULL,
+  state VARCHAR(20) NOT NULL DEFAULT 'staging',
+  backend VARCHAR(16) NOT NULL,
+  storage_endpoint_id UUID REFERENCES storage_endpoints(id) ON DELETE RESTRICT,
+  object_key TEXT NOT NULL,
+  storage_etag TEXT,
+  size_bytes BIGINT,
+  lease_owner UUID,
+  lease_until TIMESTAMPTZ,
+  delete_attempts INTEGER NOT NULL DEFAULT 0,
+  next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  last_error TEXT,
+  published_at TIMESTAMPTZ,
+  deleted_at TIMESTAMPTZ,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  CONSTRAINT storage_objects_kind_check
+    CHECK (object_kind IN ('result', 'zip_attempt', 'input', 'input_dir', 'legacy_zip')),
+  CONSTRAINT storage_objects_state_check
+    CHECK (state IN ('staging', 'published', 'delete_pending', 'deleted')),
+  CONSTRAINT storage_objects_backend_check
+    CHECK (backend IN ('s3', 'local', 'local_dir')),
+  CONSTRAINT storage_objects_attempts_check
+    CHECK (delete_attempts >= 0),
+  CONSTRAINT storage_objects_size_check
+    CHECK (size_bytes IS NULL OR size_bytes >= 0),
+  CONSTRAINT storage_objects_endpoint_check
+    CHECK (
+      (backend = 's3' AND storage_endpoint_id IS NOT NULL)
+      OR (backend IN ('local', 'local_dir') AND storage_endpoint_id IS NULL)
+    ),
+  CONSTRAINT storage_objects_lease_pair_check
+    CHECK ((lease_owner IS NULL) = (lease_until IS NULL))
+);
+
+CREATE UNIQUE INDEX storage_objects_locator_unique
+  ON storage_objects(
+    backend,
+    COALESCE(storage_endpoint_id, '00000000-0000-0000-0000-000000000000'::uuid),
+    object_key
+  );
+
+CREATE INDEX storage_objects_cleanup_ready
+  ON storage_objects(next_attempt_at, created_at)
+  WHERE state IN ('staging', 'delete_pending');
+
+CREATE INDEX storage_objects_task
+  ON storage_objects(task_id, state);
+
+INSERT INTO storage_objects (
+  task_id, task_file_id, object_kind, state,
+  backend, storage_endpoint_id, object_key, storage_etag,
+  size_bytes, published_at
+)
+SELECT
+  file.task_id,
+  file.id,
+  'result',
+  'published',
+  file.storage_backend,
+  file.storage_endpoint_id,
+  COALESCE(file.storage_key, file.storage_path),
+  file.storage_etag,
+  file.compressed_size,
+  COALESCE(file.completed_at, file.created_at)
+FROM task_files AS file
+WHERE file.status = 'completed'
+  AND COALESCE(file.storage_key, file.storage_path) IS NOT NULL
+ON CONFLICT DO NOTHING;
+
+INSERT INTO storage_objects (
+  task_id, object_kind, state,
+  backend, storage_endpoint_id, object_key, storage_etag,
+  size_bytes, published_at
+)
+SELECT
+  task.id,
+  'zip_attempt',
+  'published',
+  task.zip_storage_backend,
+  task.zip_storage_endpoint_id,
+  task.zip_storage_key,
+  task.zip_storage_etag,
+  task.zip_size,
+  COALESCE(task.completed_at, task.created_at)
+FROM tasks AS task
+WHERE task.zip_storage_backend IS NOT NULL
+  AND task.zip_storage_key IS NOT NULL
+ON CONFLICT DO NOTHING;

migrations/026_idempotency_operation_leases.sql

@@ -0,0 +1,11 @@
+ALTER TABLE idempotency_keys
+  ADD COLUMN lease_owner UUID,
+  ADD COLUMN lease_until TIMESTAMPTZ;
+
+ALTER TABLE idempotency_keys
+  ADD CONSTRAINT idempotency_keys_lease_pair_check
+  CHECK ((lease_owner IS NULL) = (lease_until IS NULL));
+
+CREATE INDEX idempotency_keys_stale_operations
+  ON idempotency_keys(lease_until)
+  WHERE response_status = 0;

migrations/027_task_target_size.sql

@@ -0,0 +1,6 @@
+ALTER TABLE tasks
+  ADD COLUMN target_size_bytes BIGINT;
+
+ALTER TABLE tasks
+  ADD CONSTRAINT tasks_target_size_bytes_check
+  CHECK (target_size_bytes IS NULL OR target_size_bytes >= 1024);

scripts/run_external_state_tests.sh

@@ -1,7 +1,7 @@
 #!/usr/bin/env bash
 set -Eeuo pipefail
 
-EXPECTED_EXTERNAL_TESTS="${EXPECTED_EXTERNAL_TESTS:-9}"
+EXPECTED_EXTERNAL_TESTS="${EXPECTED_EXTERNAL_TESTS:-13}"
 WORK_DIR="$(mktemp -d)"
 POSTGRES_CONTAINER=""
 REDIS_CONTAINER=""
@@ -79,19 +79,37 @@ case "$arch" in
   *) echo "Unsupported MinIO architecture: ${arch}" >&2; exit 1 ;;
 esac
 
-MINIO_BIN="${MINIO_BIN:-${WORK_DIR}/minio}"
+default_minio_bin="${WORK_DIR}/minio"
-MC_BIN="${MC_BIN:-${WORK_DIR}/mc}"
+default_mc_bin="${WORK_DIR}/mc"
+if command -v minio >/dev/null 2>&1; then
+  default_minio_bin="$(command -v minio)"
+fi
+if command -v mc >/dev/null 2>&1; then
+  default_mc_bin="$(command -v mc)"
+fi
+MINIO_BIN="${MINIO_BIN:-${default_minio_bin}}"
+MC_BIN="${MC_BIN:-${default_mc_bin}}"
+
+download_tool() {
+  local url=$1
+  local output=$2
+  local partial="${output}.part"
+  curl --fail --silent --show-error --location \
+    --retry 10 --retry-all-errors --retry-delay 2 --connect-timeout 20 \
+    --continue-at - --output "$partial" "$url"
+  mv "$partial" "$output"
+  chmod +x "$output"
+}
+
 if [[ ! -x "$MINIO_BIN" ]]; then
-  curl --fail --silent --show-error --location --retry 3 \
+  download_tool \
     "https://dl.min.io/server/minio/release/linux-${minio_arch}/minio" \
-    --output "$MINIO_BIN"
+    "$MINIO_BIN"
-  chmod +x "$MINIO_BIN"
 fi
 if [[ ! -x "$MC_BIN" ]]; then
-  curl --fail --silent --show-error --location --retry 3 \
+  download_tool \
     "https://dl.min.io/client/mc/release/linux-${minio_arch}/mc" \
-    --output "$MC_BIN"
+    "$MC_BIN"
-  chmod +x "$MC_BIN"
 fi
 
 export IMAGEFORGE_TEST_S3_ENDPOINT="${IMAGEFORGE_TEST_S3_ENDPOINT:-http://127.0.0.1:19000}"

src/api/admin.rs

@@ -959,28 +959,92 @@ async fn create_manual_subscription(
         return Err(AppError::new(ErrorCode::Forbidden, "套餐不可用"));
     }
 
+    let (subscription_id, period_start, period_end) = persist_manual_subscription(
+        &state.db,
+        admin_id,
+        user_id,
+        plan.id,
+        months,
+        req.note.as_deref(),
+        ip,
+    )
+    .await?;
+
+    Ok(Json(Envelope {
+        success: true,
+        data: ManualSubscriptionResponse {
+            message: "套餐已开通".to_string(),
+            subscription_id,
+            user_id,
+            plan_id: plan.id,
+            plan_name: plan.name,
+            period_start,
+            period_end,
+            status: "active".to_string(),
+        },
+    }))
+}
+
+async fn persist_manual_subscription(
+    pool: &sqlx::PgPool,
+    admin_id: Uuid,
+    user_id: Uuid,
+    plan_id: Uuid,
+    months: i32,
+    note: Option<&str>,
+    ip: IpAddr,
+) -> Result<(Uuid, DateTime<Utc>, DateTime<Utc>), AppError> {
     let period_start = Utc::now();
     let period_end = add_months_utc8(period_start, months)?;
-
+    let mut tx = pool
-    let mut tx = state
-        .db
         .begin()
         .await
         .map_err(|err| AppError::new(ErrorCode::Internal, "开启事务失败").with_source(err))?;
 
-    let _ = sqlx::query(
+    let _: Uuid = sqlx::query_scalar("SELECT id FROM users WHERE id = $1 FOR UPDATE")
+        .bind(user_id)
+        .fetch_one(&mut *tx)
+        .await
+        .map_err(|err| AppError::new(ErrorCode::Internal, "锁定订阅用户失败").with_source(err))?;
+
+    let has_open_stripe: bool = sqlx::query_scalar(
+        r#"
+        SELECT EXISTS(
+          SELECT 1
+          FROM subscriptions
+          WHERE user_id = $1
+            AND provider = 'stripe'
+            AND status <> 'canceled'
+        )
+        "#,
+    )
+    .bind(user_id)
+    .fetch_one(&mut *tx)
+    .await
+    .map_err(|err| AppError::new(ErrorCode::Internal, "检查 Stripe 订阅失败").with_source(err))?;
+    if has_open_stripe {
+        return Err(AppError::new(
+            ErrorCode::Forbidden,
+            "用户存在未取消的 Stripe 订阅，不能直接替换为手工套餐",
+        ));
+    }
+
+    sqlx::query(
         r#"
         UPDATE subscriptions
         SET status = 'canceled',
             cancel_at_period_end = false,
             canceled_at = NOW(),
             updated_at = NOW()
-        WHERE user_id = $1 AND status IN ('active', 'trialing', 'past_due')
+        WHERE user_id = $1
+          AND provider <> 'stripe'
+          AND status IN ('active', 'trialing', 'past_due')
         "#,
     )
     .bind(user_id)
     .execute(&mut *tx)
-    .await;
+    .await
+    .map_err(|err| AppError::new(ErrorCode::Internal, "关闭原本地订阅失败").with_source(err))?;
 
     let subscription_id: Uuid = sqlx::query_scalar(
         r#"
@@ -999,7 +1063,7 @@ async fn create_manual_subscription(
         "#,
     )
     .bind(user_id)
-    .bind(plan.id)
+    .bind(plan_id)
     .bind(period_start)
     .bind(period_end)
     .fetch_one(&mut *tx)
@@ -1031,9 +1095,9 @@ async fn create_manual_subscription(
     .bind(subscription_id)
     .bind(serde_json::json!({
         "target_user_id": user_id,
-        "plan_id": plan.id,
+        "plan_id": plan_id,
         "months": months,
-        "note": req.note,
+        "note": note,
     }))
     .bind(ip.to_string())
     .execute(&mut *tx)
@@ -1044,19 +1108,7 @@ async fn create_manual_subscription(
         .await
         .map_err(|err| AppError::new(ErrorCode::Internal, "提交事务失败").with_source(err))?;
 
-    Ok(Json(Envelope {
+    Ok((subscription_id, period_start, period_end))
-        success: true,
-        data: ManualSubscriptionResponse {
-            message: "套餐已开通".to_string(),
-            subscription_id,
-            user_id,
-            plan_id: plan.id,
-            plan_name: plan.name,
-            period_start,
-            period_end,
-            status: "active".to_string(),
-        },
-    }))
 }
 
 fn add_months_utc8(start: DateTime<Utc>, months: i32) -> Result<DateTime<Utc>, AppError> {
@@ -1740,6 +1792,7 @@ async fn audit_config_action(
 #[cfg(test)]
 mod tests {
     use super::*;
+    use sqlx::postgres::PgPoolOptions;
 
     #[test]
     fn secret_masking_never_splits_utf8() {
@@ -1747,4 +1800,151 @@ mod tests {
         assert_eq!(mask_secret("中文密钥测试内容"), "中文密钥测试内容");
         assert_eq!(mask_secret("🔑🔑🔑🔑🔑🔑🔑🔑more"), "🔑🔑🔑🔑🔑🔑🔑🔑...");
     }
+
+    #[tokio::test]
+    #[ignore = "requires IMAGEFORGE_TEST_DATABASE_URL"]
+    async fn manual_subscriptions_are_serialized_and_cannot_replace_stripe() {
+        let database_url = std::env::var("IMAGEFORGE_TEST_DATABASE_URL")
+            .expect("IMAGEFORGE_TEST_DATABASE_URL is required");
+        let pool = PgPoolOptions::new()
+            .max_connections(32)
+            .connect(&database_url)
+            .await
+            .expect("connect test database");
+        sqlx::migrate!()
+            .run(&pool)
+            .await
+            .expect("apply test migrations");
+
+        let marker = Uuid::new_v4().simple().to_string();
+        let admin_id: Uuid = sqlx::query_scalar(
+            r#"
+            INSERT INTO users (email, username, password_hash, role, email_verified_at)
+            VALUES ($1, $2, 'test', 'admin', NOW())
+            RETURNING id
+            "#,
+        )
+        .bind(format!("admin-{marker}@example.test"))
+        .bind(format!("admin-{marker}"))
+        .fetch_one(&pool)
+        .await
+        .expect("insert test admin");
+        let user_id: Uuid = sqlx::query_scalar(
+            r#"
+            INSERT INTO users (email, username, password_hash, email_verified_at)
+            VALUES ($1, $2, 'test', NOW())
+            RETURNING id
+            "#,
+        )
+        .bind(format!("user-{marker}@example.test"))
+        .bind(format!("user-{marker}"))
+        .fetch_one(&pool)
+        .await
+        .expect("insert test user");
+        let plan_id: Uuid = sqlx::query_scalar("SELECT id FROM plans WHERE code = 'pro_monthly'")
+            .fetch_one(&pool)
+            .await
+            .expect("load test plan");
+        let ip: IpAddr = "127.0.0.1".parse().unwrap();
+
+        let mut joins = Vec::new();
+        for _ in 0..20 {
+            let pool = pool.clone();
+            joins.push(tokio::spawn(async move {
+                persist_manual_subscription(
+                    &pool,
+                    admin_id,
+                    user_id,
+                    plan_id,
+                    1,
+                    Some("concurrency-test"),
+                    ip,
+                )
+                .await
+            }));
+        }
+        for join in joins {
+            join.await
+                .expect("manual subscription task panicked")
+                .expect("manual subscription failed");
+        }
+
+        let effective_manual: i64 = sqlx::query_scalar(
+            r#"
+            SELECT COUNT(*)
+            FROM subscriptions
+            WHERE user_id = $1
+              AND provider = 'manual'
+              AND status IN ('active', 'trialing', 'past_due')
+            "#,
+        )
+        .bind(user_id)
+        .fetch_one(&pool)
+        .await
+        .expect("count effective manual subscriptions");
+        assert_eq!(effective_manual, 1);
+
+        sqlx::query(
+            "UPDATE subscriptions SET status = 'canceled', canceled_at = NOW() WHERE user_id = $1",
+        )
+        .bind(user_id)
+        .execute(&pool)
+        .await
+        .expect("cancel test manual subscription");
+        sqlx::query(
+            r#"
+            INSERT INTO subscriptions (
+              user_id, plan_id, status, current_period_start, current_period_end,
+              provider, provider_customer_id, provider_subscription_id
+            ) VALUES (
+              $1, $2, 'active', NOW(), NOW() + INTERVAL '1 month',
+              'stripe', $3, $4
+            )
+            "#,
+        )
+        .bind(user_id)
+        .bind(plan_id)
+        .bind(format!("cus_{marker}"))
+        .bind(format!("sub_{marker}"))
+        .execute(&pool)
+        .await
+        .expect("insert Stripe subscription");
+
+        let error = persist_manual_subscription(
+            &pool,
+            admin_id,
+            user_id,
+            plan_id,
+            1,
+            Some("must-not-replace-stripe"),
+            ip,
+        )
+        .await
+        .expect_err("manual subscription replaced Stripe");
+        assert_eq!(error.code, ErrorCode::Forbidden);
+        let effective_subscriptions: i64 = sqlx::query_scalar(
+            r#"
+            SELECT COUNT(*)
+            FROM subscriptions
+            WHERE user_id = $1
+              AND status IN ('active', 'trialing', 'past_due')
+            "#,
+        )
+        .bind(user_id)
+        .fetch_one(&pool)
+        .await
+        .expect("count effective subscriptions");
+        assert_eq!(effective_subscriptions, 1);
+
+        sqlx::query("DELETE FROM audit_logs WHERE details->>'target_user_id' = $1")
+            .bind(user_id.to_string())
+            .execute(&pool)
+            .await
+            .expect("clean test audit logs");
+        sqlx::query("DELETE FROM users WHERE id = ANY($1)")
+            .bind(vec![user_id, admin_id])
+            .execute(&pool)
+            .await
+            .expect("clean test users");
+    }
 }

src/api/billing.rs

@@ -457,23 +457,30 @@ async fn create_checkout_for_user(
     .map_err(|err| AppError::new(ErrorCode::Internal, "锁定用户失败").with_source(err))?
     .ok_or_else(|| AppError::new(ErrorCode::Unauthorized, "用户不存在"))?;
 
-    let has_open_subscription: bool = sqlx::query_scalar(
+    let open_subscription_provider: Option<String> = sqlx::query_scalar(
         r#"
-        SELECT EXISTS(
+        SELECT provider
-          SELECT 1 FROM subscriptions
+        FROM subscriptions
-          WHERE user_id = $1 AND provider = 'stripe' AND status <> 'canceled'
+        WHERE user_id = $1
+          AND (
+            (provider = 'stripe' AND status <> 'canceled')
+            OR status IN ('active', 'trialing', 'past_due')
           )
+        ORDER BY CASE WHEN provider = 'stripe' THEN 0 ELSE 1 END
+        LIMIT 1
         "#,
     )
     .bind(user_id)
-    .fetch_one(&mut *tx)
+    .fetch_optional(&mut *tx)
     .await
     .map_err(|err| AppError::new(ErrorCode::Internal, "查询订阅状态失败").with_source(err))?;
-    if has_open_subscription {
+    if let Some(provider) = open_subscription_provider {
-        return Err(AppError::new(
+        let message = if provider == "stripe" {
-            ErrorCode::IdempotencyConflict,
+            "已有 Stripe 订阅，请通过账单门户升级、降级或续费"
-            "已有 Stripe 订阅，请通过账单门户升级、降级或续费",
+        } else {
-        ));
+            "当前已有有效套餐，请在套餐结束后创建 Stripe 订阅"
+        };
+        return Err(AppError::new(ErrorCode::IdempotencyConflict, message));
     }
 
     sqlx::query(

src/api/compress.rs

@@ -8,6 +8,7 @@ use crate::services::compress;
 use crate::services::compress::{CompressionLevel, ImageFmt};
 use crate::services::filename;
 use crate::services::idempotency;
+use crate::services::object_lifecycle;
 use crate::services::quota;
 use crate::services::storage;
 use crate::state::AppState;
@@ -21,6 +22,7 @@ use chrono::{DateTime, Duration, Utc};
 use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
 use sqlx::FromRow;
+use std::future::Future;
 use std::net::{IpAddr, SocketAddr};
 use uuid::Uuid;
 
@@ -30,6 +32,14 @@ pub fn router() -> Router<AppState> {
         .route("/compress/direct", post(compress_direct))
 }
 
+fn spawn_detached_operation<F, T>(future: F) -> tokio::task::JoinHandle<T>
+where
+    F: Future<Output = T> + Send + 'static,
+    T: Send + 'static,
+{
+    tokio::spawn(future)
+}
+
 #[derive(Debug, Serialize, Deserialize)]
 struct BillingView {
     units_charged: i32,
@@ -214,6 +224,7 @@ async fn compress_json(
     let quota_ctx = admission.quota_ctx;
 
     let mut idem_acquired = false;
+    let mut idem_owner = None;
     if let (Some(scope), Some(idem_key), Some(request_hash)) = (
         idempotency_scope,
         idempotency_key.as_deref(),
@@ -262,13 +273,33 @@ async fn compress_json(
                     "请求正在处理中，请稍后重试",
                 ));
             }
-            idempotency::BeginResult::Acquired => {
+            idempotency::BeginResult::Acquired { owner } => {
                 idem_acquired = true;
+                idem_owner = Some(owner);
             }
         }
     }
 
     let task_id = Uuid::new_v4();
+    let operation_state = state.clone();
+    let operation_principal = principal.clone();
+    let operation_quota_ctx = quota_ctx.clone();
+    let operation_idempotency_key = idempotency_key.clone();
+    let operation_request_hash = request_hash.clone();
+    let operation_idem_owner = idem_owner;
+    let operation = spawn_detached_operation(async move {
+        let state = operation_state;
+        let principal = operation_principal;
+        let quota_ctx = operation_quota_ctx;
+        let idempotency_key = operation_idempotency_key;
+        let request_hash = operation_request_hash;
+        let _idempotency_heartbeat = start_idempotency_heartbeat(
+            &state,
+            idempotency_scope,
+            idempotency_key.as_deref(),
+            request_hash.as_deref(),
+            operation_idem_owner,
+        );
         let mut anonymous_reservation_date = None;
         let op: Result<CompressResponse, AppError> = (async {
         match &quota_ctx {
@@ -318,60 +349,19 @@ async fn compress_json(
         let retention_hours = retention.num_hours();
         let object_key =
             storage::result_key(retention_hours, task_id, file_id, format_out.extension());
-        let stored =
+        let tracked = object_lifecycle::store_tracked_bytes(
-            storage::store_bytes(&state, &object_key, compressed, format_out.content_type())
+            &state,
+            task_id,
+            Some(file_id),
+            "result",
+            &object_key,
+            compressed,
+            format_out.content_type(),
+        )
         .await?;
 
         let expires_at = Utc::now() + retention;
-
+        let response = CompressResponse {
-        if let Err(err) = record_task_and_metering(
-            &state,
-            &principal,
-            ip,
-            task_id,
-            file_id,
-            &stored,
-            &req.file_name,
-            req.max_width,
-            req.max_height,
-            effective_level,
-            req.compression_rate,
-            format_in,
-            format_out,
-            original_size,
-            compressed_size,
-            saved_percent,
-            expires_at,
-            retention_hours,
-            &quota_ctx,
-            charge_units,
-        )
-        .await
-        {
-            let _ = storage::delete_object(
-                &state,
-                &storage::ObjectLocator {
-                    backend: stored.backend.clone(),
-                    endpoint_id: stored.endpoint_id,
-                    key: stored.key.clone(),
-                },
-            )
-            .await;
-            return Err(err);
-        }
-
-        if anonymous_reservation_date.is_some() {
-            if let Err(err) =
-                quota::finalize_anonymous_single_reservation(&state, task_id, charge_units).await
-            {
-                // The durable reservation remains visible to maintenance, so a
-                // transient Redis failure must not turn a successful image into
-                // a failed, non-idempotent request.
-                tracing::warn!(task_id = %task_id, charged = charge_units, error = %err, "anonymous single reservation finalization deferred");
-            }
-        }
-
-        Ok(CompressResponse {
             task_id,
             file_id,
             format_in: format_in.as_str().to_string(),
@@ -385,37 +375,93 @@ async fn compress_json(
             billing: BillingView {
                 units_charged: metered_units(charge_units),
             },
-        })
+        };
-    })
+        let idem_completion = build_idempotency_completion(
-    .await;
+            idem_acquired,
-
+            operation_idem_owner,
-    match op {
-        Ok(resp) => {
-            if let (Some(scope), Some(idem_key), Some(request_hash)) = (
             idempotency_scope,
             idempotency_key.as_deref(),
             request_hash.as_deref(),
-            ) {
+            &response,
-                if idem_acquired {
+        )?;
-                    let _ = idempotency::complete(
+
+        if let Err(err) = record_task_and_metering(
             &state,
-                        scope,
+            &principal,
-                        idem_key,
+            ip,
-                        request_hash,
+            task_id,
-                        200,
+            file_id,
-                        serde_json::to_value(&resp).unwrap_or(serde_json::Value::Null),
+            &tracked,
+            &req.file_name,
+            req.max_width,
+            req.max_height,
+            effective_level,
+            req.compression_rate,
+            req.target_size_bytes,
+            format_in,
+            format_out,
+            original_size,
+            compressed_size,
+            saved_percent,
+            expires_at,
+            retention_hours,
+            &quota_ctx,
+            charge_units,
+            idem_completion.as_ref(),
         )
+        .await
+        {
+            match sync_result_was_committed(&state, task_id, file_id, &tracked).await {
+                Ok(true) => {
+                    tracing::warn!(task_id = %task_id, file_id = %file_id, error = %err, "sync result commit response was lost; recovered committed publication");
+                }
+                Ok(false) => {
+                    if let Err(cleanup_err) = object_lifecycle::schedule_tracked_delete(
+                        &state,
+                        &tracked,
+                        Some(&err),
+                    )
+                    .await
+                    {
+                        tracing::error!(task_id = %task_id, storage_object_id = %tracked.lifecycle_id, error = %cleanup_err, "failed to persist rejected sync result cleanup");
+                    }
+                    return Err(err);
+                }
+                Err(probe_err) => {
+                    tracing::error!(task_id = %task_id, storage_object_id = %tracked.lifecycle_id, error = %probe_err, original_error = %err, "sync result commit state is unknown; staging lease will reconcile object");
+                    return Err(err);
+                }
+            }
+        }
+
+        if anonymous_reservation_date.is_some() {
+            if let Err(err) =
+                quota::finalize_anonymous_single_reservation(&state, task_id, charge_units).await
+            {
+                // The durable reservation remains visible to maintenance, so a
+                // transient Redis failure must not turn a successful image into
+                // a failed, non-idempotent request.
+                tracing::warn!(task_id = %task_id, charged = charge_units, error = %err, "anonymous single reservation finalization deferred");
+            }
+        }
+
+        Ok(response)
+        })
         .await;
-                }
+        (op, anonymous_reservation_date)
-            }
+    });
-            Ok((
+    let (op, anonymous_reservation_date) = operation.await.map_err(|err| {
+        AppError::new(ErrorCode::Internal, "同步压缩后台任务异常退出").with_source(err)
+    })?;
+
+    match op {
+        Ok(resp) => Ok((
             jar,
             Json(Envelope {
                 success: true,
                 data: resp,
             }),
-            ))
+        )),
-        }
         Err(err) => {
             if anonymous_reservation_date.is_some() {
                 if let Err(refund_err) =
@@ -430,7 +476,10 @@ async fn compress_json(
                 request_hash.as_deref(),
             ) {
                 if idem_acquired {
-                    let _ = idempotency::abort(&state, scope, idem_key, request_hash).await;
+                    if let Some(owner) = idem_owner {
+                        let _ =
+                            idempotency::abort(&state, scope, idem_key, request_hash, owner).await;
+                    }
                 }
             }
             Err(err)
@@ -525,6 +574,7 @@ async fn compress_direct(
     let quota_ctx = admission.quota_ctx;
 
     let mut idem_acquired = false;
+    let mut idem_owner = None;
     if let (Some(scope), Some(idem_key), Some(request_hash)) = (
         idempotency_scope,
         idempotency_key.as_deref(),
@@ -566,12 +616,32 @@ async fn compress_direct(
                     "请求正在处理中，请稍后重试",
                 ));
             }
-            idempotency::BeginResult::Acquired => {
+            idempotency::BeginResult::Acquired { owner } => {
                 idem_acquired = true;
+                idem_owner = Some(owner);
             }
         }
     }
 
+    let operation_state = state.clone();
+    let operation_principal = principal.clone();
+    let operation_quota_ctx = quota_ctx.clone();
+    let operation_idempotency_key = idempotency_key.clone();
+    let operation_request_hash = request_hash.clone();
+    let operation_idem_owner = idem_owner;
+    let operation = spawn_detached_operation(async move {
+        let state = operation_state;
+        let principal = operation_principal;
+        let quota_ctx = operation_quota_ctx;
+        let idempotency_key = operation_idempotency_key;
+        let request_hash = operation_request_hash;
+        let _idempotency_heartbeat = start_idempotency_heartbeat(
+            &state,
+            idempotency_scope,
+            idempotency_key.as_deref(),
+            request_hash.as_deref(),
+            operation_idem_owner,
+        );
         let op: Result<(axum::response::Response, DirectIdempotencyData), AppError> = (async {
         match &quota_ctx {
             QuotaContext::User(billing) => ensure_quota_available(&state, billing, 1).await?,
@@ -617,8 +687,11 @@ async fn compress_direct(
         let retention_hours = retention.num_hours();
         let object_key =
             storage::result_key(retention_hours, task_id, file_id, format_out.extension());
-        let stored = storage::store_bytes(
+        let tracked = object_lifecycle::store_tracked_bytes(
             &state,
+            task_id,
+            Some(file_id),
+            "result",
             &object_key,
             compressed.clone(),
             format_out.content_type(),
@@ -626,43 +699,6 @@ async fn compress_direct(
         .await?;
 
         let expires_at = Utc::now() + retention;
-
-        if let Err(err) = record_task_and_metering(
-            &state,
-            &principal,
-            ip,
-            task_id,
-            file_id,
-            &stored,
-            &req.file_name,
-            req.max_width,
-            req.max_height,
-            effective_level,
-            req.compression_rate,
-            format_in,
-            format_out,
-            original_size,
-            compressed_size,
-            saved_percent,
-            expires_at,
-            retention_hours,
-            &quota_ctx,
-            charge_units,
-        )
-        .await
-        {
-            let _ = storage::delete_object(
-                &state,
-                &storage::ObjectLocator {
-                    backend: stored.backend.clone(),
-                    endpoint_id: stored.endpoint_id,
-                    key: stored.key.clone(),
-                },
-            )
-            .await;
-            return Err(err);
-        }
-
         let idem_data = DirectIdempotencyData {
             file_id,
             format_out: format_out.as_str().to_string(),
@@ -672,32 +708,76 @@ async fn compress_direct(
             saved_percent,
             units_charged: metered_units(charge_units),
         };
+        let idem_completion = build_idempotency_completion(
+            idem_acquired,
+            operation_idem_owner,
+            idempotency_scope,
+            idempotency_key.as_deref(),
+            request_hash.as_deref(),
+            &idem_data,
+        )?;
+
+        if let Err(err) = record_task_and_metering(
+            &state,
+            &principal,
+            ip,
+            task_id,
+            file_id,
+            &tracked,
+            &req.file_name,
+            req.max_width,
+            req.max_height,
+            effective_level,
+            req.compression_rate,
+            req.target_size_bytes,
+            format_in,
+            format_out,
+            original_size,
+            compressed_size,
+            saved_percent,
+            expires_at,
+            retention_hours,
+            &quota_ctx,
+            charge_units,
+            idem_completion.as_ref(),
+        )
+        .await
+        {
+            match sync_result_was_committed(&state, task_id, file_id, &tracked).await {
+                Ok(true) => {
+                    tracing::warn!(task_id = %task_id, file_id = %file_id, error = %err, "direct result commit response was lost; recovered committed publication");
+                }
+                Ok(false) => {
+                    if let Err(cleanup_err) = object_lifecycle::schedule_tracked_delete(
+                        &state,
+                        &tracked,
+                        Some(&err),
+                    )
+                    .await
+                    {
+                        tracing::error!(task_id = %task_id, storage_object_id = %tracked.lifecycle_id, error = %cleanup_err, "failed to persist rejected direct result cleanup");
+                    }
+                    return Err(err);
+                }
+                Err(probe_err) => {
+                    tracing::error!(task_id = %task_id, storage_object_id = %tracked.lifecycle_id, error = %probe_err, original_error = %err, "direct result commit state is unknown; staging lease will reconcile object");
+                    return Err(err);
+                }
+            }
+        }
+
         let response = direct_response(compressed, format_out, &idem_data);
         Ok((response, idem_data))
         })
         .await;
+        op
+    });
+    let op = operation.await.map_err(|err| {
+        AppError::new(ErrorCode::Internal, "直接压缩后台任务异常退出").with_source(err)
+    })?;
 
     match op {
-        Ok((response, idem_data)) => {
+        Ok((response, _idem_data)) => Ok((jar, response)),
-            if let (Some(scope), Some(idem_key), Some(request_hash)) = (
-                idempotency_scope,
-                idempotency_key.as_deref(),
-                request_hash.as_deref(),
-            ) {
-                if idem_acquired {
-                    let _ = idempotency::complete(
-                        &state,
-                        scope,
-                        idem_key,
-                        request_hash,
-                        200,
-                        serde_json::to_value(&idem_data).unwrap_or(serde_json::Value::Null),
-                    )
-                    .await;
-                }
-            }
-            Ok((jar, response))
-        }
         Err(err) => {
             if let (Some(scope), Some(idem_key), Some(request_hash)) = (
                 idempotency_scope,
@@ -705,7 +785,10 @@ async fn compress_direct(
                 request_hash.as_deref(),
             ) {
                 if idem_acquired {
-                    let _ = idempotency::abort(&state, scope, idem_key, request_hash).await;
+                    if let Some(owner) = idem_owner {
+                        let _ =
+                            idempotency::abort(&state, scope, idem_key, request_hash, owner).await;
+                    }
                 }
             }
             Err(err)
@@ -713,6 +796,45 @@ async fn compress_direct(
     }
 }
 
+async fn sync_result_was_committed(
+    state: &AppState,
+    task_id: Uuid,
+    file_id: Uuid,
+    tracked: &object_lifecycle::TrackedStoredObject,
+) -> Result<bool, AppError> {
+    sqlx::query_scalar(
+        r#"
+        SELECT EXISTS(
+          SELECT 1
+          FROM tasks AS task
+          JOIN task_files AS file ON file.task_id = task.id
+          JOIN storage_objects AS object ON object.id = $3
+          WHERE task.id = $1
+            AND task.status = 'completed'
+            AND file.id = $2
+            AND file.status = 'completed'
+            AND file.storage_backend = $4
+            AND file.storage_endpoint_id IS NOT DISTINCT FROM $5
+            AND COALESCE(file.storage_key, file.storage_path) = $6
+            AND object.state = 'published'
+            AND object.task_id = task.id
+            AND object.task_file_id = file.id
+        )
+        "#,
+    )
+    .bind(task_id)
+    .bind(file_id)
+    .bind(tracked.lifecycle_id)
+    .bind(&tracked.stored.backend)
+    .bind(tracked.stored.endpoint_id)
+    .bind(&tracked.stored.key)
+    .fetch_one(&state.db)
+    .await
+    .map_err(|err| {
+        AppError::new(ErrorCode::StorageUnavailable, "核验同步压缩提交结果失败").with_source(err)
+    })
+}
+
 #[derive(Debug, FromRow)]
 struct DirectReplayRow {
     storage_backend: String,
@@ -742,6 +864,7 @@ async fn load_direct_replay_bytes(
                 FROM task_files f
                 JOIN tasks t ON t.id = f.task_id
                 WHERE f.id = $1 AND t.user_id = $2
+                  AND t.deletion_started_at IS NULL
                 "#,
             )
             .bind(file_id)
@@ -762,6 +885,7 @@ async fn load_direct_replay_bytes(
                 FROM task_files f
                 JOIN tasks t ON t.id = f.task_id
                 WHERE f.id = $1 AND t.api_key_id = $2
+                  AND t.deletion_started_at IS NULL
                 "#,
             )
             .bind(file_id)
@@ -900,7 +1024,6 @@ async fn parse_single_file_request(
                             "target_size_bytes 格式错误，需为正整数（字节）",
                         )
                     })?);
-                    // 最小目标大小限制：1KB
                     if let Some(size) = target_size_bytes {
                         if size < 1024 {
                             return Err(AppError::new(
@@ -908,6 +1031,12 @@ async fn parse_single_file_request(
                                 "target_size_bytes 最小为 1024（1KB）",
                             ));
                         }
+                        if i64::try_from(size).is_err() {
+                            return Err(AppError::new(
+                                ErrorCode::InvalidRequest,
+                                "target_size_bytes 超出支持范围",
+                            ));
+                        }
                     }
                 }
             }
@@ -946,6 +1075,63 @@ enum QuotaContext {
     ApiKey(BillingContext, Uuid),
 }
 
+struct IdempotencyCompletion {
+    scope: idempotency::Scope,
+    owner: Uuid,
+    key: String,
+    request_hash: String,
+    response_body: serde_json::Value,
+}
+
+fn build_idempotency_completion<T: Serialize>(
+    acquired: bool,
+    owner: Option<Uuid>,
+    scope: Option<idempotency::Scope>,
+    key: Option<&str>,
+    request_hash: Option<&str>,
+    response: &T,
+) -> Result<Option<IdempotencyCompletion>, AppError> {
+    if !acquired {
+        return Ok(None);
+    }
+    let (owner, scope, key, request_hash) = match (owner, scope, key, request_hash) {
+        (Some(owner), Some(scope), Some(key), Some(request_hash)) => {
+            (owner, scope, key, request_hash)
+        }
+        _ => return Err(AppError::new(ErrorCode::Internal, "幂等请求上下文不完整")),
+    };
+    Ok(Some(IdempotencyCompletion {
+        scope,
+        owner,
+        key: key.to_string(),
+        request_hash: request_hash.to_string(),
+        response_body: serde_json::to_value(response).map_err(|err| {
+            AppError::new(ErrorCode::Internal, "序列化幂等响应失败").with_source(err)
+        })?,
+    }))
+}
+
+fn start_idempotency_heartbeat(
+    state: &AppState,
+    scope: Option<idempotency::Scope>,
+    key: Option<&str>,
+    request_hash: Option<&str>,
+    owner: Option<Uuid>,
+) -> Option<idempotency::LeaseHeartbeat> {
+    match (scope, key, request_hash, owner) {
+        (Some(scope), Some(key), Some(request_hash), Some(owner)) => {
+            Some(idempotency::start_lease_heartbeat(
+                state.clone(),
+                scope,
+                key.to_string(),
+                request_hash.to_string(),
+                owner,
+            ))
+        }
+        _ => None,
+    }
+}
+
 struct SingleAdmission {
     retention: Duration,
     quota_ctx: QuotaContext,
@@ -1044,12 +1230,13 @@ async fn record_task_and_metering(
     client_ip: IpAddr,
     task_id: Uuid,
     file_id: Uuid,
-    stored: &storage::StoredObject,
+    tracked: &object_lifecycle::TrackedStoredObject,
     original_name: &str,
     max_width: Option<u32>,
     max_height: Option<u32>,
     level: CompressionLevel,
     compression_rate: Option<u8>,
+    target_size_bytes: Option<u64>,
     format_in: ImageFmt,
     format_out: ImageFmt,
     original_size: u64,
@@ -1059,7 +1246,9 @@ async fn record_task_and_metering(
     retention_hours: i64,
     quota_ctx: &QuotaContext,
     charge_units: bool,
+    idempotency_completion: Option<&IdempotencyCompletion>,
 ) -> Result<(), AppError> {
+    let stored = &tracked.stored;
     let (user_id, session_id, api_key_id, source) = match principal {
         context::Principal::Anonymous { session_id } => {
             (None, Some(session_id.clone()), None, "web")
@@ -1083,16 +1272,16 @@ async fn record_task_and_metering(
         INSERT INTO tasks (
           id, user_id, session_id, api_key_id, client_ip, source, status,
           compression_level, output_format, max_width, max_height, preserve_metadata,
-          compression_rate,
+          compression_rate, target_size_bytes,
           total_files, completed_files, failed_files,
           total_original_size, total_compressed_size,
           started_at, completed_at, expires_at, retention_hours
         ) VALUES (
           $1, $2, $3, $4, $5::inet, $6::task_source, 'completed',
           $7::compression_level, $8, $9, $10, $11, $12,
-          1, 1, 0,
+          $13, 1, 1, 0,
-          $13, $14,
+          $14, $15,
-          NOW(), NOW(), $15, $16
+          NOW(), NOW(), $16, $17
         )
         "#,
     )
@@ -1108,6 +1297,7 @@ async fn record_task_and_metering(
     .bind(max_height.map(|v| v as i32))
     .bind(false)
     .bind(compression_rate.map(|v| v as i16))
+    .bind(target_size_bytes.map(|v| v as i64))
     .bind(original_size as i64)
     .bind(compressed_size as i64)
     .bind(expires_at)
@@ -1191,6 +1381,20 @@ async fn record_task_and_metering(
         }
     }
 
+    object_lifecycle::publish_in_tx(&mut tx, tracked).await?;
+    if let Some(idempotency_completion) = idempotency_completion {
+        idempotency::complete_in_tx(
+            &mut tx,
+            idempotency_completion.scope,
+            &idempotency_completion.key,
+            &idempotency_completion.request_hash,
+            idempotency_completion.owner,
+            200,
+            idempotency_completion.response_body.clone(),
+        )
+        .await?;
+    }
+
     tx.commit()
         .await
         .map_err(|err| AppError::new(ErrorCode::Internal, "提交事务失败").with_source(err))?;
@@ -1244,6 +1448,8 @@ async fn charge_one_unit(
 #[cfg(test)]
 mod tests {
     use super::*;
+    use std::sync::atomic::{AtomicBool, Ordering};
+    use std::sync::Arc;
 
     #[test]
     fn direct_response_has_consistent_compression_headers() {
@@ -1296,4 +1502,35 @@ mod tests {
             assert_eq!(metered_units(charged), expected_units);
         }
     }
+
+    #[tokio::test]
+    async fn detached_sync_operation_survives_waiter_abort() {
+        let started = Arc::new(tokio::sync::Notify::new());
+        let release = Arc::new(tokio::sync::Notify::new());
+        let completed = Arc::new(AtomicBool::new(false));
+        let request_started = started.clone();
+        let request_release = release.clone();
+        let request_completed = completed.clone();
+        let request = tokio::spawn(async move {
+            spawn_detached_operation(async move {
+                request_started.notify_one();
+                request_release.notified().await;
+                request_completed.store(true, Ordering::SeqCst);
+            })
+            .await
+            .expect("detached operation panicked");
+        });
+
+        started.notified().await;
+        request.abort();
+        request.await.expect_err("request waiter was not aborted");
+        release.notify_one();
+        tokio::time::timeout(std::time::Duration::from_secs(1), async {
+            while !completed.load(Ordering::SeqCst) {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("detached operation was canceled with its request waiter");
+    }
 }

src/api/downloads.rs

@@ -1,5 +1,6 @@
 use crate::api::context;
 use crate::error::{AppError, ErrorCode};
+use crate::services::object_lifecycle;
 use crate::services::storage;
 use crate::state::AppState;
 
@@ -15,6 +16,7 @@ use sqlx::FromRow;
 use std::collections::HashMap;
 use std::net::SocketAddr;
 use std::path::PathBuf;
+use tokio::sync::oneshot;
 use tokio_util::io::ReaderStream;
 use uuid::Uuid;
 
@@ -62,7 +64,7 @@ async fn download_file(
           t.expires_at
         FROM task_files f
         JOIN tasks t ON t.id = f.task_id
-        WHERE f.id = $1
+        WHERE f.id = $1 AND t.deletion_started_at IS NULL
         "#,
     )
     .bind(file_id)
@@ -234,7 +236,7 @@ struct TaskZipRow {
     zip_storage_key: Option<String>,
 }
 
-#[derive(Debug, FromRow)]
+#[derive(Clone, Debug, FromRow)]
 struct TaskZipFileRow {
     storage_backend: String,
     storage_endpoint_id: Option<Uuid>,
@@ -259,8 +261,10 @@ enum ZipBuildClaim {
     Busy,
 }
 
-const ZIP_BUILD_LEASE_SECONDS: i64 = 15 * 60;
+const ZIP_BUILD_LEASE_SECONDS: i64 = 60;
-const ZIP_BUILD_WAIT_SECONDS: u64 = 30;
+const ZIP_BUILD_HEARTBEAT_SECONDS: u64 = 20;
+const ZIP_BUILD_WAIT_SECONDS: u64 = 300;
+const ZIP_BUILD_GATE_WAIT_SECONDS: u64 = 60;
 
 async fn download_task_zip(
     State(state): State<AppState>,
@@ -286,7 +290,7 @@ async fn download_task_zip(
           zip_storage_endpoint_id,
           zip_storage_key
         FROM tasks
-        WHERE id = $1
+        WHERE id = $1 AND deletion_started_at IS NULL
         "#,
     )
     .bind(task_id)
@@ -380,7 +384,16 @@ async fn resolve_task_zip(
         match claim_zip_build(state, task_id).await? {
             ZipBuildClaim::Cached(object) => return Ok(object),
             ZipBuildClaim::Acquired { token } => {
-                return build_claimed_zip(state, task_id, retention_hours, token, &rows).await;
+                let receiver = spawn_claimed_zip_build(
+                    state.clone(),
+                    task_id,
+                    retention_hours,
+                    token,
+                    rows.clone(),
+                );
+                return receiver.await.map_err(|err| {
+                    AppError::new(ErrorCode::Internal, "后台 ZIP 构建任务异常退出").with_source(err)
+                })?;
             }
             ZipBuildClaim::Busy if tokio::time::Instant::now() < deadline => {
                 tokio::time::sleep(std::time::Duration::from_millis(100)).await;
@@ -439,6 +452,7 @@ async fn claim_zip_build(state: &AppState, task_id: Uuid) -> Result<ZipBuildClai
             zip_build_attempt = zip_build_attempt + 1
         WHERE id = $1
           AND zip_storage_key IS NULL
+          AND deletion_started_at IS NULL
           AND completed_at IS NOT NULL
           AND expires_at > NOW()
           AND (
@@ -462,7 +476,7 @@ async fn claim_zip_build(state: &AppState, task_id: Uuid) -> Result<ZipBuildClai
         r#"
         SELECT zip_storage_backend, zip_storage_endpoint_id, zip_storage_key, expires_at
         FROM tasks
-        WHERE id = $1
+        WHERE id = $1 AND deletion_started_at IS NULL
         "#,
     )
     .bind(task_id)
@@ -486,6 +500,15 @@ async fn claim_zip_build(state: &AppState, task_id: Uuid) -> Result<ZipBuildClai
 }
 
 async fn renew_zip_build(state: &AppState, task_id: Uuid, token: Uuid) -> Result<(), AppError> {
+    renew_zip_build_for(state, task_id, token, ZIP_BUILD_LEASE_SECONDS).await
+}
+
+async fn renew_zip_build_for(
+    state: &AppState,
+    task_id: Uuid,
+    token: Uuid,
+    lease_seconds: i64,
+) -> Result<(), AppError> {
     let updated = sqlx::query(
         r#"
         UPDATE tasks
@@ -494,11 +517,12 @@ async fn renew_zip_build(state: &AppState, task_id: Uuid, token: Uuid) -> Result
           AND zip_build_token = $2
           AND zip_storage_key IS NULL
           AND expires_at > NOW()
+          AND deletion_started_at IS NULL
         "#,
     )
     .bind(task_id)
     .bind(token)
-    .bind(ZIP_BUILD_LEASE_SECONDS)
+    .bind(lease_seconds)
     .execute(&state.db)
     .await
     .map_err(|err| AppError::new(ErrorCode::Internal, "续租 ZIP 构建失败").with_source(err))?;
@@ -511,6 +535,98 @@ async fn renew_zip_build(state: &AppState, task_id: Uuid, token: Uuid) -> Result
     Ok(())
 }
 
+struct ZipHeartbeatStop(Option<oneshot::Sender<()>>);
+
+impl Drop for ZipHeartbeatStop {
+    fn drop(&mut self) {
+        if let Some(stop) = self.0.take() {
+            let _ = stop.send(());
+        }
+    }
+}
+
+fn start_zip_build_heartbeat(
+    state: AppState,
+    task_id: Uuid,
+    token: Uuid,
+    interval: std::time::Duration,
+    lease_seconds: i64,
+) -> (ZipHeartbeatStop, tokio::task::JoinHandle<()>) {
+    let (stop_tx, mut stop_rx) = oneshot::channel();
+    let handle = tokio::spawn(async move {
+        let mut ticker = tokio::time::interval(interval);
+        ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+        ticker.tick().await;
+        loop {
+            tokio::select! {
+                _ = &mut stop_rx => break,
+                _ = ticker.tick() => {
+                    if let Err(err) = renew_zip_build_for(
+                        &state,
+                        task_id,
+                        token,
+                        lease_seconds,
+                    )
+                    .await
+                    {
+                        tracing::warn!(task_id = %task_id, zip_build_token = %token, error = %err, "failed to heartbeat ZIP build lease");
+                    }
+                }
+            }
+        }
+    });
+    (ZipHeartbeatStop(Some(stop_tx)), handle)
+}
+
+fn spawn_claimed_zip_build(
+    state: AppState,
+    task_id: Uuid,
+    retention_hours: i64,
+    token: Uuid,
+    rows: Vec<TaskZipFileRow>,
+) -> oneshot::Receiver<Result<storage::ObjectLocator, AppError>> {
+    let (result_tx, result_rx) = oneshot::channel();
+    tokio::spawn(async move {
+        let result = run_claimed_zip_build(state, task_id, retention_hours, token, rows).await;
+        let _ = result_tx.send(result);
+    });
+    result_rx
+}
+
+async fn run_claimed_zip_build(
+    state: AppState,
+    task_id: Uuid,
+    retention_hours: i64,
+    token: Uuid,
+    rows: Vec<TaskZipFileRow>,
+) -> Result<storage::ObjectLocator, AppError> {
+    let (heartbeat_stop, heartbeat) = start_zip_build_heartbeat(
+        state.clone(),
+        task_id,
+        token,
+        std::time::Duration::from_secs(ZIP_BUILD_HEARTBEAT_SECONDS),
+        ZIP_BUILD_LEASE_SECONDS,
+    );
+    let build_state = state.clone();
+    let build = tokio::spawn(async move {
+        build_claimed_zip(&build_state, task_id, retention_hours, token, &rows).await
+    });
+    let result = match build.await {
+        Ok(result) => result,
+        Err(err) => {
+            let temp_dir = zip_temp_dir(&state, task_id, token);
+            let _ = tokio::fs::remove_dir_all(temp_dir).await;
+            release_zip_build(&state, task_id, token).await;
+            Err(AppError::new(ErrorCode::Internal, "ZIP 构建任务异常退出").with_source(err))
+        }
+    };
+    drop(heartbeat_stop);
+    if let Err(err) = heartbeat.await {
+        tracing::warn!(task_id = %task_id, zip_build_token = %token, error = %err, "ZIP lease heartbeat task failed");
+    }
+    result
+}
+
 async fn release_zip_build(state: &AppState, task_id: Uuid, token: Uuid) {
     if let Err(err) = sqlx::query(
         r#"
@@ -539,7 +655,7 @@ async fn build_claimed_zip(
     rows: &[TaskZipFileRow],
 ) -> Result<storage::ObjectLocator, AppError> {
     let permit = match tokio::time::timeout(
-        std::time::Duration::from_secs(ZIP_BUILD_WAIT_SECONDS),
+        std::time::Duration::from_secs(ZIP_BUILD_GATE_WAIT_SECONDS),
         state.zip_build_semaphore.clone().acquire_owned(),
     )
     .await
@@ -558,10 +674,7 @@ async fn build_claimed_zip(
         }
     };
 
-    let temp_dir = PathBuf::from(format!(
+    let temp_dir = zip_temp_dir(state, task_id, token);
-        "{}/tmp/zips/{task_id}-{token}",
-        state.config.storage_path
-    ));
     let zip_path = temp_dir.join(format!("task_{task_id}.zip"));
     let build_result = build_zip_attempt(
         state,
@@ -586,6 +699,13 @@ async fn build_claimed_zip(
     publish_zip_attempt(state, task_id, token, stored).await
 }
 
+fn zip_temp_dir(state: &AppState, task_id: Uuid, token: Uuid) -> PathBuf {
+    PathBuf::from(&state.config.storage_path)
+        .join("tmp")
+        .join("zips")
+        .join(format!("{task_id}-{token}"))
+}
+
 async fn build_zip_attempt(
     state: &AppState,
     task_id: Uuid,
@@ -594,7 +714,7 @@ async fn build_zip_attempt(
     rows: &[TaskZipFileRow],
     temp_dir: &std::path::Path,
     zip_path: &std::path::Path,
-) -> Result<storage::StoredObject, AppError> {
+) -> Result<object_lifecycle::TrackedStoredObject, AppError> {
     tokio::fs::create_dir_all(temp_dir).await.map_err(|err| {
         AppError::new(ErrorCode::StorageUnavailable, "创建 ZIP 临时目录失败").with_source(err)
     })?;
@@ -649,15 +769,27 @@ async fn build_zip_attempt(
 
     renew_zip_build(state, task_id, token).await?;
     let object_key = storage::archive_attempt_key(retention_hours, task_id, token);
-    storage::store_file(state, &object_key, zip_path, "application/zip").await
+    object_lifecycle::store_tracked_file(
+        state,
+        task_id,
+        "zip_attempt",
+        &object_key,
+        zip_path,
+        "application/zip",
+    )
+    .await
 }
 
 async fn publish_zip_attempt(
     state: &AppState,
     task_id: Uuid,
     token: Uuid,
-    stored: storage::StoredObject,
+    tracked: object_lifecycle::TrackedStoredObject,
 ) -> Result<storage::ObjectLocator, AppError> {
+    let stored = &tracked.stored;
+    let mut tx = state.db.begin().await.map_err(|err| {
+        AppError::new(ErrorCode::Internal, "开启 ZIP 发布事务失败").with_source(err)
+    })?;
     let published = sqlx::query(
         r#"
         UPDATE tasks
@@ -672,6 +804,7 @@ async fn publish_zip_attempt(
           AND zip_build_token = $2
           AND zip_storage_key IS NULL
           AND expires_at > NOW()
+          AND deletion_started_at IS NULL
         "#,
     )
     .bind(task_id)
@@ -681,24 +814,36 @@ async fn publish_zip_attempt(
     .bind(&stored.key)
     .bind(&stored.etag)
     .bind(stored.size as i64)
-    .execute(&state.db)
+    .execute(&mut *tx)
     .await;
 
     match published {
-        Ok(result) if result.rows_affected() == 1 => Ok(storage::ObjectLocator {
+        Ok(result) if result.rows_affected() == 1 => {
-            backend: stored.backend,
+            if let Err(err) = object_lifecycle::publish_in_tx(&mut tx, &tracked).await {
-            endpoint_id: stored.endpoint_id,
+                tx.rollback().await.ok();
-            key: stored.key,
+                delete_unpublished_zip(state, task_id, token, &tracked).await;
-        }),
+                return Err(err);
+            }
+            tx.commit().await.map_err(|err| {
+                AppError::new(ErrorCode::Internal, "提交 ZIP 发布事务失败").with_source(err)
+            })?;
+            Ok(storage::ObjectLocator {
+                backend: tracked.stored.backend,
+                endpoint_id: tracked.stored.endpoint_id,
+                key: tracked.stored.key,
+            })
+        }
         Ok(_) => {
-            delete_unpublished_zip(state, task_id, token, &stored).await;
+            tx.rollback().await.ok();
+            delete_unpublished_zip(state, task_id, token, &tracked).await;
             let current = load_published_zip(state, task_id).await?;
             current.ok_or_else(|| {
                 AppError::new(ErrorCode::StorageUnavailable, "ZIP 发布租约已失效，请重试")
             })
         }
         Err(err) => {
-            delete_unpublished_zip(state, task_id, token, &stored).await;
+            tx.rollback().await.ok();
+            delete_unpublished_zip(state, task_id, token, &tracked).await;
             Err(AppError::new(ErrorCode::Internal, "记录 ZIP 对象失败").with_source(err))
         }
     }
@@ -708,28 +853,13 @@ async fn delete_unpublished_zip(
     state: &AppState,
     task_id: Uuid,
     token: Uuid,
-    stored: &storage::StoredObject,
+    tracked: &object_lifecycle::TrackedStoredObject,
 ) {
-    let object = storage::ObjectLocator {
+    if let Err(err) = object_lifecycle::schedule_tracked_delete(state, tracked, None).await {
-        backend: stored.backend.clone(),
+        tracing::error!(task_id = %task_id, zip_build_token = %token, object_key = %tracked.stored.key, error = %err, "failed to persist unpublished ZIP deletion");
-        endpoint_id: stored.endpoint_id,
+    } else if let Err(err) = object_lifecycle::cleanup_ready_objects(state, 1, Some(task_id)).await
-        key: stored.key.clone(),
+    {
-    };
+        tracing::warn!(task_id = %task_id, zip_build_token = %token, object_key = %tracked.stored.key, error = %err, "unpublished ZIP deletion deferred");
-    let mut last_error = None;
-    for attempt in 1..=3_u64 {
-        match storage::delete_object(state, &object).await {
-            Ok(()) => {
-                last_error = None;
-                break;
-            }
-            Err(err) => {
-                last_error = Some(err);
-                tokio::time::sleep(std::time::Duration::from_millis(100 * attempt)).await;
-            }
-        }
-    }
-    if let Some(err) = last_error {
-        tracing::error!(task_id = %task_id, zip_build_token = %token, object_key = %stored.key, error = %err, "failed to delete unpublished ZIP attempt after retries");
     }
     release_zip_build(state, task_id, token).await;
 }
@@ -739,7 +869,7 @@ async fn load_published_zip(
     task_id: Uuid,
 ) -> Result<Option<storage::ObjectLocator>, AppError> {
     let row: Option<(Option<String>, Option<Uuid>, Option<String>)> = sqlx::query_as(
-        "SELECT zip_storage_backend, zip_storage_endpoint_id, zip_storage_key FROM tasks WHERE id = $1",
+        "SELECT zip_storage_backend, zip_storage_endpoint_id, zip_storage_key FROM tasks WHERE id = $1 AND deletion_started_at IS NULL",
     )
     .bind(task_id)
     .fetch_optional(&state.db)
@@ -1130,8 +1260,10 @@ mod tests {
         tokio::fs::write(&unpublished_path, b"unpublished-zip")
             .await
             .expect("write unpublished ZIP fixture");
-        let unpublished = storage::store_file(
+        let unpublished = object_lifecycle::store_tracked_file(
             &state,
+            deleted_task,
+            "zip_attempt",
             &storage::archive_attempt_key(24, deleted_task, deleted_token),
             &unpublished_path,
             "application/zip",
@@ -1151,24 +1283,174 @@ mod tests {
         let orphan_read = storage::read_bytes(
             &state,
             &storage::ObjectLocator {
-                backend: unpublished.backend,
+                backend: unpublished.stored.backend,
-                endpoint_id: unpublished.endpoint_id,
+                endpoint_id: unpublished.stored.endpoint_id,
-                key: unpublished.key,
+                key: unpublished.stored.key,
             },
         )
         .await;
         assert!(orphan_read.is_err(), "unpublished ZIP object was orphaned");
+        let unpublished_state: String =
+            sqlx::query_scalar("SELECT state FROM storage_objects WHERE id = $1")
+                .bind(unpublished.lifecycle_id)
+                .fetch_one(&pool)
+                .await
+                .expect("query unpublished ZIP lifecycle state");
+        assert_eq!(unpublished_state, "deleted");
 
-        for locator in [&locators[0], &takeover_locator] {
+        let cancelled_task = Uuid::new_v4();
+        insert_zip_task(
+            &pool,
+            cancelled_task,
+            &format!("{marker}-cancelled-request"),
+            &input_path,
+            21,
+        )
+        .await;
+        let gate_one = state
+            .zip_build_semaphore
+            .clone()
+            .acquire_owned()
+            .await
+            .expect("acquire first ZIP gate permit");
+        let gate_two = state
+            .zip_build_semaphore
+            .clone()
+            .acquire_owned()
+            .await
+            .expect("acquire second ZIP gate permit");
+        let cancelled_state = state.clone();
+        let cancelled_request =
+            tokio::spawn(
+                async move { resolve_task_zip(&cancelled_state, cancelled_task, 24).await },
+            );
+        let mut cancelled_token = None;
+        for _ in 0..100 {
+            cancelled_token = sqlx::query_scalar("SELECT zip_build_token FROM tasks WHERE id = $1")
+                .bind(cancelled_task)
+                .fetch_one(&pool)
+                .await
+                .expect("query cancelled-request ZIP token");
+            if cancelled_token.is_some() {
+                break;
+            }
+            tokio::time::sleep(std::time::Duration::from_millis(25)).await;
+        }
+        let cancelled_token = cancelled_token.expect("background ZIP builder never acquired lease");
+        cancelled_request.abort();
+        assert!(
+            cancelled_request
+                .await
+                .expect_err("aborted ZIP request completed")
+                .is_cancelled(),
+            "ZIP request was not cancelled"
+        );
+        drop(gate_one);
+        drop(gate_two);
+
+        let mut cancelled_locator = None;
+        for _ in 0..200 {
+            cancelled_locator = load_published_zip(&state, cancelled_task)
+                .await
+                .expect("query background ZIP publication");
+            if cancelled_locator.is_some() {
+                break;
+            }
+            tokio::time::sleep(std::time::Duration::from_millis(25)).await;
+        }
+        let cancelled_locator =
+            cancelled_locator.expect("request cancellation stopped the background ZIP build");
+        let cancelled_state_row: (Option<Uuid>, i64) =
+            sqlx::query_as("SELECT zip_build_token, zip_build_attempt FROM tasks WHERE id = $1")
+                .bind(cancelled_task)
+                .fetch_one(&pool)
+                .await
+                .expect("query background ZIP terminal state");
+        assert_eq!(cancelled_state_row, (None, 1));
+        assert!(
+            !tokio::fs::try_exists(zip_temp_dir(&state, cancelled_task, cancelled_token))
+                .await
+                .expect("check cancelled-request ZIP temp directory"),
+            "background ZIP temp directory survived publication"
+        );
+
+        let heartbeat_task = Uuid::new_v4();
+        insert_zip_task(
+            &pool,
+            heartbeat_task,
+            &format!("{marker}-heartbeat"),
+            &input_path,
+            21,
+        )
+        .await;
+        let heartbeat_token = match claim_zip_build(&state, heartbeat_task)
+            .await
+            .expect("claim heartbeat ZIP builder")
+        {
+            ZipBuildClaim::Acquired { token } => token,
+            other => panic!("unexpected heartbeat ZIP claim: {other:?}"),
+        };
+        sqlx::query(
+            "UPDATE tasks SET zip_build_lease_until = NOW() + INTERVAL '1 second' WHERE id = $1 AND zip_build_token = $2",
+        )
+        .bind(heartbeat_task)
+        .bind(heartbeat_token)
+        .execute(&pool)
+        .await
+        .expect("shorten heartbeat ZIP lease");
+        let (heartbeat_stop, heartbeat) = start_zip_build_heartbeat(
+            state.clone(),
+            heartbeat_task,
+            heartbeat_token,
+            std::time::Duration::from_millis(100),
+            1,
+        );
+        tokio::time::sleep(std::time::Duration::from_millis(1_500)).await;
+        assert!(
+            matches!(
+                claim_zip_build(&state, heartbeat_task)
+                    .await
+                    .expect("probe heartbeat ZIP lease"),
+                ZipBuildClaim::Busy
+            ),
+            "heartbeat did not fence a takeover after the original lease duration"
+        );
+        drop(heartbeat_stop);
+        heartbeat.await.expect("join ZIP heartbeat task");
+        release_zip_build(&state, heartbeat_task, heartbeat_token).await;
+
+        for locator in [&locators[0], &takeover_locator, &cancelled_locator] {
             storage::delete_object(&state, locator)
                 .await
                 .expect("delete published ZIP test object");
         }
         sqlx::query("DELETE FROM tasks WHERE id = ANY($1)")
-            .bind(&[task_id, over_budget_task, takeover_task][..])
+            .bind(
+                &[
+                    task_id,
+                    over_budget_task,
+                    takeover_task,
+                    cancelled_task,
+                    heartbeat_task,
+                ][..],
+            )
             .execute(&pool)
             .await
             .expect("delete ZIP test tasks");
+        sqlx::query("DELETE FROM storage_objects WHERE task_id = ANY($1)")
+            .bind(
+                &[
+                    task_id,
+                    over_budget_task,
+                    takeover_task,
+                    deleted_task,
+                    cancelled_task,
+                    heartbeat_task,
+                ][..],
+            )
+            .execute(&pool)
+            .await
+            .expect("delete ZIP lifecycle test rows");
         if let Some(endpoint_id) = endpoint_id {
             sqlx::query("DELETE FROM storage_endpoints WHERE id = $1")
                 .bind(endpoint_id)

src/api/metrics.rs

@@ -17,17 +17,32 @@ const SCRAPE_TIMEOUT: Duration = Duration::from_secs(2);
 pub async fn metrics(State(state): State<AppState>) -> impl IntoResponse {
     let database = tokio::time::timeout(
         SCRAPE_TIMEOUT,
-        sqlx::query_scalar::<_, i64>(
+        sqlx::query_as::<_, (i64, i64, i64, i64, i64)>(
-            "SELECT COUNT(*) FROM tasks WHERE status IN ('pending', 'processing')",
+            r#"
+            SELECT
+              (SELECT COUNT(*) FROM tasks WHERE status IN ('pending', 'processing')),
+              (SELECT COUNT(*) FROM task_queue_outbox WHERE status IN ('pending', 'delivering')),
+              (SELECT COUNT(*) FROM task_queue_outbox WHERE status = 'dead'),
+              (SELECT COUNT(*) FROM storage_objects WHERE state = 'delete_pending'),
+              (SELECT COUNT(*) FROM storage_objects WHERE state = 'staging')
+            "#,
         )
         .fetch_one(&state.db),
     );
     let redis = tokio::time::timeout(SCRAPE_TIMEOUT, redis_queue_stats(state.redis.clone()));
     let (database, redis) = tokio::join!(database, redis);
 
-    let (database_up, active_tasks) = match database {
+    let (database_up, active_tasks, outbox_pending, outbox_dead, delete_pending, staging) =
-        Ok(Ok(value)) => (1, value),
+        match database {
-        _ => (0, 0),
+            Ok(Ok((active, outbox_pending, outbox_dead, delete_pending, staging))) => (
+                1,
+                active,
+                outbox_pending,
+                outbox_dead,
+                delete_pending,
+                staging,
+            ),
+            _ => (0, 0, 0, 0, 0, 0),
         };
     let (redis_up, queue_length, pending, dead_length, cluster) = match redis {
         Ok(Ok((queue_length, pending, dead_length, cluster))) => {
@@ -51,6 +66,28 @@ pub async fn metrics(State(state): State<AppState>) -> impl IntoResponse {
     output.push_str("# HELP imageforge_active_tasks Current pending or processing tasks.\n");
     output.push_str("# TYPE imageforge_active_tasks gauge\n");
     let _ = writeln!(output, "imageforge_active_tasks {active_tasks}");
+    output.push_str("# HELP imageforge_task_outbox Current durable task delivery states.\n");
+    output.push_str("# TYPE imageforge_task_outbox gauge\n");
+    let _ = writeln!(
+        output,
+        "imageforge_task_outbox{{state=\"pending\"}} {outbox_pending}"
+    );
+    let _ = writeln!(
+        output,
+        "imageforge_task_outbox{{state=\"dead\"}} {outbox_dead}"
+    );
+    output.push_str(
+        "# HELP imageforge_storage_object_lifecycle Current durable object cleanup states.\n",
+    );
+    output.push_str("# TYPE imageforge_storage_object_lifecycle gauge\n");
+    let _ = writeln!(
+        output,
+        "imageforge_storage_object_lifecycle{{state=\"delete_pending\"}} {delete_pending}"
+    );
+    let _ = writeln!(
+        output,
+        "imageforge_storage_object_lifecycle{{state=\"staging\"}} {staging}"
+    );
     output.push_str("# HELP imageforge_queue_messages Current Redis stream message counts.\n");
     output.push_str("# TYPE imageforge_queue_messages gauge\n");
     let _ = writeln!(

src/api/mod.rs

@@ -53,12 +53,19 @@ pub async fn run(state: AppState) -> Result<(), AppError> {
     tracing::info!(addr = %addr, "API server listening");
 
     let reconciliation_task = tokio::spawn(webhooks::reconciliation_loop(state.clone()));
+    let queue_dispatch_task =
+        tokio::spawn(crate::services::task_queue::dispatch_loop(state.clone()));
+    let object_lifecycle_task = tokio::spawn(crate::services::object_lifecycle::maintenance_loop(
+        state.clone(),
+    ));
     let serve_result = axum::serve(
         listener,
         app.into_make_service_with_connect_info::<SocketAddr>(),
     )
     .await;
     reconciliation_task.abort();
+    queue_dispatch_task.abort();
+    object_lifecycle_task.abort();
     serve_result
         .map_err(|err| AppError::new(ErrorCode::Internal, "HTTP 服务异常退出").with_source(err))
 }

src/api/webhooks.rs

@@ -642,12 +642,14 @@ async fn write_subscription(
     if subscription.status != "canceled" {
         let conflicting: Option<String> = sqlx::query_scalar(
             r#"
-            SELECT provider_subscription_id
+            SELECT provider || ':' || COALESCE(provider_subscription_id, id::text)
             FROM subscriptions
             WHERE user_id = $1
-              AND provider = 'stripe'
+              AND status IN ('active', 'trialing', 'past_due')
-              AND status <> 'canceled'
+              AND NOT (
-              AND provider_subscription_id <> $2
+                provider = 'stripe'
+                AND provider_subscription_id = $2
+              )
             FOR UPDATE
             "#,
         )
@@ -661,7 +663,7 @@ async fn write_subscription(
         if conflicting.is_some() {
             return Err(AppError::new(
                 ErrorCode::StorageUnavailable,
-                "用户已有其他未取消 Stripe 订阅，事件等待人工对账",
+                "用户已有其他有效订阅，Stripe 事件等待人工对账",
             ));
         }
     }
@@ -1910,6 +1912,42 @@ mod tests {
             .await
             .expect("cancel primary subscription");
 
+        let manual_subscription_id: Uuid = sqlx::query_scalar(
+            r#"
+            INSERT INTO subscriptions (
+              user_id, plan_id, status, current_period_start, current_period_end, provider
+            ) VALUES ($1, $2, 'active', NOW(), NOW() + INTERVAL '1 month', 'manual')
+            RETURNING id
+            "#,
+        )
+        .bind(user_id)
+        .bind(plan_id)
+        .fetch_one(&pool)
+        .await
+        .expect("insert manual subscription before delayed Stripe event");
+        let delayed_active_error = apply_test_event(&state, &secondary)
+            .await
+            .expect_err("delayed active Stripe event created cross-provider double entitlement");
+        assert_eq!(delayed_active_error.code, ErrorCode::StorageUnavailable);
+        let effective_after_delay: i64 = sqlx::query_scalar(
+            r#"
+            SELECT COUNT(*) FROM subscriptions
+            WHERE user_id = $1 AND status IN ('active', 'trialing', 'past_due')
+            "#,
+        )
+        .bind(user_id)
+        .fetch_one(&pool)
+        .await
+        .expect("count effective subscriptions after delayed Stripe event");
+        assert_eq!(effective_after_delay, 1);
+        sqlx::query(
+            "UPDATE subscriptions SET status = 'canceled', canceled_at = NOW() WHERE id = $1",
+        )
+        .bind(manual_subscription_id)
+        .execute(&pool)
+        .await
+        .expect("cancel delayed-event manual fixture");
+
         let concurrent_invoice_id = format!("inv_{marker}_concurrent");
         let concurrent_invoice_number = format!("INV-{marker}-C");
         let concurrent_invoice_events = [

src/services/compress.rs

@@ -33,11 +33,12 @@ const AVIF_TARGET_MIN_QUALITY: u8 = 38;
 const JPEG_PERCEPTUAL_QUALITY: u8 = 72;
 const WEBP_PERCEPTUAL_QUALITY: u8 = 70;
 const AVIF_PERCEPTUAL_QUALITY: u8 = 55;
-const JPEG_TARGET_MAX_QUALITY: u8 = 90;
+const JPEG_TARGET_MAX_QUALITY: u8 = 100;
-const WEBP_TARGET_MAX_QUALITY: u8 = 92;
+const WEBP_TARGET_MAX_QUALITY: u8 = 100;
-const AVIF_TARGET_MAX_QUALITY: u8 = 90;
+const AVIF_TARGET_MAX_QUALITY: u8 = 100;
 const AVIF_ENCODER_SPEED: u8 = 5;
 const WEBP_TARGET_SAFETY_PERCENT: u64 = 97;
+const WEBP_HIGH_EFFORT_LOSSLESS_MAX_PIXELS: u64 = 2_100_000;
 const METADATA_TARGET_OVERHEAD: u64 = 1024;
 
 #[derive(Clone)]
@@ -927,6 +928,23 @@ fn encode_webp_target(
 ) -> Result<Vec<u8>, AppError> {
     deadline.check()?;
     let pixels = prepare_target_pixels(&image);
+    let lossless_candidate = encode_webp_lossless_pixels(&pixels);
+    deadline.check()?;
+    match lossless_candidate {
+        Ok(lossless) if lossless.len() as u64 <= target_size => return Ok(lossless),
+        Ok(_) => {}
+        Err(error) => {
+            tracing::debug!(error = %error, "WebP 无损候选编码失败，继续尝试有损编码");
+        }
+    }
+    deadline.check()?;
+
+    let max_lossy = encode_webp_pixels(&pixels, WEBP_TARGET_MAX_QUALITY)?;
+    deadline.check()?;
+    if max_lossy.len() as u64 <= target_size {
+        return Ok(max_lossy);
+    }
+
     let native_min_quality = if allow_resize {
         WEBP_PERCEPTUAL_QUALITY
     } else {
@@ -987,6 +1005,37 @@ fn encode_webp_native_target(
         })
 }
 
+fn encode_webp_lossless_pixels(pixels: &TargetPixels) -> Result<Vec<u8>, AppError> {
+    let mut config = webp::WebPConfig::new()
+        .map_err(|_| AppError::new(ErrorCode::CompressionFailed, "初始化 WebP 无损配置失败"))?;
+    config.lossless = 1;
+    config.quality = 100.0;
+    config.method = webp_lossless_method(pixels);
+    config.alpha_compression = 1;
+    config.near_lossless = 100;
+    config.exact = 1;
+    config.thread_level = 0;
+
+    webp_encoder(pixels)
+        .encode_advanced(&config)
+        .map(|bytes| bytes.to_vec())
+        .map_err(|err| {
+            AppError::new(
+                ErrorCode::CompressionFailed,
+                format!("WebP 无损编码失败: {err:?}"),
+            )
+        })
+}
+
+fn webp_lossless_method(pixels: &TargetPixels) -> i32 {
+    let pixel_count = u64::from(pixels.width).saturating_mul(u64::from(pixels.height));
+    if pixel_count <= WEBP_HIGH_EFFORT_LOSSLESS_MAX_PIXELS {
+        6
+    } else {
+        0
+    }
+}
+
 fn encode_avif_target(
     image: DynamicImage,
     target_size: u64,
@@ -1942,6 +1991,79 @@ mod tests {
         assert_eq!(detect_format(&output).unwrap(), ImageFmt::Webp);
     }
 
+    #[test]
+    fn webp_target_prefers_lossless_when_it_fits() {
+        let image = DynamicImage::ImageRgb8(RgbImage::from_fn(160, 120, |x, y| {
+            let block = ((x / 20) + (y / 20) * 3) as u8;
+            Rgb([
+                block.wrapping_mul(31),
+                block.wrapping_mul(17),
+                block.wrapping_mul(11),
+            ])
+        }));
+        let pixels = prepare_target_pixels(&image);
+        let lossless = encode_webp_lossless_pixels(&pixels).unwrap();
+        let output = encode_webp_target(
+            image.clone(),
+            lossless.len() as u64,
+            true,
+            &CompressionDeadline::unlimited(),
+        )
+        .unwrap();
+
+        assert_eq!(output, lossless);
+        assert_eq!(
+            image::load_from_memory(&output).unwrap().to_rgb8(),
+            image.to_rgb8()
+        );
+    }
+
+    #[test]
+    fn webp_target_uses_quality_100_when_lossless_exceeds_the_cap() {
+        let mut state = 0x7f4a_7c15_u32;
+        let image = DynamicImage::ImageRgb8(RgbImage::from_fn(160, 120, |_x, _y| {
+            let mut channel = || {
+                state ^= state << 13;
+                state ^= state >> 17;
+                state ^= state << 5;
+                state as u8
+            };
+            Rgb([channel(), channel(), channel()])
+        }));
+        let pixels = prepare_target_pixels(&image);
+        let max_lossy = encode_webp_pixels(&pixels, WEBP_TARGET_MAX_QUALITY).unwrap();
+        let lossless = encode_webp_lossless_pixels(&pixels).unwrap();
+        assert!(max_lossy.len() < lossless.len());
+
+        let output = encode_webp_target(
+            image,
+            max_lossy.len() as u64,
+            true,
+            &CompressionDeadline::unlimited(),
+        )
+        .unwrap();
+        assert_eq!(output, max_lossy);
+    }
+
+    #[test]
+    fn large_webp_targets_use_the_fast_lossless_probe() {
+        let small = TargetPixels {
+            bytes: Vec::new(),
+            width: 1920,
+            height: 1080,
+            layout: TargetPixelLayout::Rgb,
+        };
+        let large = TargetPixels {
+            bytes: Vec::new(),
+            width: 4096,
+            height: 3072,
+            layout: TargetPixelLayout::Rgb,
+        };
+
+        assert_eq!(webp_lossless_method(&small), 6);
+        assert_eq!(webp_lossless_method(&large), 0);
+    }
+
     #[test]
     fn jpeg_target_encoder_prefers_perceptual_downscaling() {
         let image = DynamicImage::ImageRgb8(RgbImage::from_fn(800, 600, |x, y| {

src/services/idempotency.rs

@@ -6,6 +6,8 @@ use serde_json::Value as JsonValue;
 use sqlx::FromRow;
 use uuid::Uuid;
 
+const OPERATION_LEASE_MINUTES: i64 = 30;
+
 #[derive(Debug, Clone, Copy)]
 pub enum Scope {
     User(Uuid),
@@ -14,7 +16,7 @@ pub enum Scope {
 
 #[derive(Debug)]
 pub enum BeginResult {
-    Acquired,
+    Acquired { owner: Uuid },
     Replay { response_body: JsonValue },
     InProgress,
 }
@@ -54,6 +56,8 @@ pub async fn begin(
 
     let now = Utc::now();
     let expires_at = now + Duration::hours(ttl_hours.max(1));
+    let owner = Uuid::new_v4();
+    let lease_until = now + Duration::minutes(OPERATION_LEASE_MINUTES);
 
     cleanup_expired_for_key(state, scope, idempotency_key, now).await?;
 
@@ -64,11 +68,11 @@ pub async fn begin(
                 INSERT INTO idempotency_keys (
                   user_id, idempotency_key, request_hash,
                   response_status, response_body,
-                  expires_at
+                  expires_at, lease_owner, lease_until
                 ) VALUES (
                   $1, $2, $3,
                   0, NULL,
-                  $4
+                  $4, $5, $6
                 )
                 ON CONFLICT DO NOTHING
                 "#,
@@ -77,6 +81,8 @@ pub async fn begin(
             .bind(idempotency_key)
             .bind(request_hash)
             .bind(expires_at)
+            .bind(owner)
+            .bind(lease_until)
             .execute(&state.db)
             .await
         }
@@ -86,11 +92,11 @@ pub async fn begin(
                 INSERT INTO idempotency_keys (
                   api_key_id, idempotency_key, request_hash,
                   response_status, response_body,
-                  expires_at
+                  expires_at, lease_owner, lease_until
                 ) VALUES (
                   $1, $2, $3,
                   0, NULL,
-                  $4
+                  $4, $5, $6
                 )
                 ON CONFLICT DO NOTHING
                 "#,
@@ -99,6 +105,8 @@ pub async fn begin(
             .bind(idempotency_key)
             .bind(request_hash)
             .bind(expires_at)
+            .bind(owner)
+            .bind(lease_until)
             .execute(&state.db)
             .await
         }
@@ -106,12 +114,15 @@ pub async fn begin(
     .map_err(|err| AppError::new(ErrorCode::Internal, "写入幂等记录失败").with_source(err))?;
 
     if inserted.rows_affected() > 0 {
-        return Ok(BeginResult::Acquired);
+        return Ok(BeginResult::Acquired { owner });
     }
 
     let row = get_row(state, scope, idempotency_key, now).await?;
     let Some(row) = row else {
-        return Ok(BeginResult::Acquired);
+        return Err(AppError::new(
+            ErrorCode::StorageUnavailable,
+            "幂等记录状态已变化，请重试",
+        ));
     };
 
     if row.request_hash != request_hash {
@@ -122,6 +133,19 @@ pub async fn begin(
     }
 
     if row.response_status == 0 || row.response_body.is_none() {
+        if take_over_stale_operation(
+            state,
+            scope,
+            idempotency_key,
+            request_hash,
+            owner,
+            lease_until,
+            now,
+        )
+        .await?
+        {
+            return Ok(BeginResult::Acquired { owner });
+        }
         return Ok(BeginResult::InProgress);
     }
 
@@ -130,6 +154,152 @@ pub async fn begin(
     })
 }
 
+async fn take_over_stale_operation(
+    state: &AppState,
+    scope: Scope,
+    idempotency_key: &str,
+    request_hash: &str,
+    owner: Uuid,
+    lease_until: DateTime<Utc>,
+    now: DateTime<Utc>,
+) -> Result<bool, AppError> {
+    let updated = match scope {
+        Scope::User(user_id) => {
+            sqlx::query(
+                r#"
+                UPDATE idempotency_keys
+                SET lease_owner = $4, lease_until = $5
+                WHERE user_id = $1 AND idempotency_key = $2
+                  AND request_hash = $3 AND response_status = 0
+                  AND (lease_until IS NULL OR lease_until <= $6)
+                "#,
+            )
+            .bind(user_id)
+            .bind(idempotency_key)
+            .bind(request_hash)
+            .bind(owner)
+            .bind(lease_until)
+            .bind(now)
+            .execute(&state.db)
+            .await
+        }
+        Scope::ApiKey(api_key_id) => {
+            sqlx::query(
+                r#"
+                UPDATE idempotency_keys
+                SET lease_owner = $4, lease_until = $5
+                WHERE api_key_id = $1 AND idempotency_key = $2
+                  AND request_hash = $3 AND response_status = 0
+                  AND (lease_until IS NULL OR lease_until <= $6)
+                "#,
+            )
+            .bind(api_key_id)
+            .bind(idempotency_key)
+            .bind(request_hash)
+            .bind(owner)
+            .bind(lease_until)
+            .bind(now)
+            .execute(&state.db)
+            .await
+        }
+    }
+    .map_err(|err| AppError::new(ErrorCode::Internal, "接管过期幂等操作失败").with_source(err))?;
+    Ok(updated.rows_affected() == 1)
+}
+
+pub struct LeaseHeartbeat(Option<tokio::sync::oneshot::Sender<()>>);
+
+impl Drop for LeaseHeartbeat {
+    fn drop(&mut self) {
+        if let Some(stop) = self.0.take() {
+            let _ = stop.send(());
+        }
+    }
+}
+
+pub fn start_lease_heartbeat(
+    state: AppState,
+    scope: Scope,
+    idempotency_key: String,
+    request_hash: String,
+    owner: Uuid,
+) -> LeaseHeartbeat {
+    let (stop_tx, mut stop_rx) = tokio::sync::oneshot::channel();
+    tokio::spawn(async move {
+        let mut interval = tokio::time::interval(std::time::Duration::from_secs(60));
+        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+        interval.tick().await;
+        loop {
+            tokio::select! {
+                _ = &mut stop_rx => break,
+                _ = interval.tick() => {
+                    match renew_lease(
+                        &state,
+                        scope,
+                        &idempotency_key,
+                        &request_hash,
+                        owner,
+                    ).await {
+                        Ok(true) => {}
+                        Ok(false) => break,
+                        Err(err) => tracing::warn!(error = %err, "failed to renew idempotency operation lease"),
+                    }
+                }
+            }
+        }
+    });
+    LeaseHeartbeat(Some(stop_tx))
+}
+
+async fn renew_lease(
+    state: &AppState,
+    scope: Scope,
+    idempotency_key: &str,
+    request_hash: &str,
+    owner: Uuid,
+) -> Result<bool, AppError> {
+    let updated = match scope {
+        Scope::User(user_id) => {
+            sqlx::query(
+                r#"
+                UPDATE idempotency_keys
+                SET lease_until = NOW() + ($5 * INTERVAL '1 minute')
+                WHERE user_id = $1 AND idempotency_key = $2
+                  AND request_hash = $3 AND lease_owner = $4
+                  AND response_status = 0
+                "#,
+            )
+            .bind(user_id)
+            .bind(idempotency_key)
+            .bind(request_hash)
+            .bind(owner)
+            .bind(OPERATION_LEASE_MINUTES)
+            .execute(&state.db)
+            .await
+        }
+        Scope::ApiKey(api_key_id) => {
+            sqlx::query(
+                r#"
+                UPDATE idempotency_keys
+                SET lease_until = NOW() + ($5 * INTERVAL '1 minute')
+                WHERE api_key_id = $1 AND idempotency_key = $2
+                  AND request_hash = $3 AND lease_owner = $4
+                  AND response_status = 0
+                "#,
+            )
+            .bind(api_key_id)
+            .bind(idempotency_key)
+            .bind(request_hash)
+            .bind(owner)
+            .bind(OPERATION_LEASE_MINUTES)
+            .execute(&state.db)
+            .await
+        }
+    }
+    .map_err(|err| AppError::new(ErrorCode::Internal, "续租幂等操作失败").with_source(err))?;
+    Ok(updated.rows_affected() == 1)
+}
+
 pub async fn wait_for_replay(
     state: &AppState,
     scope: Scope,
@@ -165,11 +335,12 @@ pub async fn wait_for_replay(
     }
 }
 
-pub async fn complete(
+pub async fn complete_in_tx(
-    state: &AppState,
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
     scope: Scope,
     idempotency_key: &str,
     request_hash: &str,
+    owner: Uuid,
     response_status: i32,
     response_body: JsonValue,
 ) -> Result<(), AppError> {
@@ -179,10 +350,13 @@ pub async fn complete(
                 r#"
                 UPDATE idempotency_keys
                 SET response_status = $4,
-                    response_body = $5
+                    response_body = $5,
+                    lease_owner = NULL,
+                    lease_until = NULL
                 WHERE user_id = $1
                   AND idempotency_key = $2
                   AND request_hash = $3
+                  AND lease_owner = $6
                   AND response_status = 0
                 "#,
             )
@@ -191,7 +365,8 @@ pub async fn complete(
             .bind(request_hash)
             .bind(response_status)
             .bind(response_body)
-            .execute(&state.db)
+            .bind(owner)
+            .execute(&mut **tx)
             .await
         }
         Scope::ApiKey(api_key_id) => {
@@ -199,10 +374,13 @@ pub async fn complete(
                 r#"
                 UPDATE idempotency_keys
                 SET response_status = $4,
-                    response_body = $5
+                    response_body = $5,
+                    lease_owner = NULL,
+                    lease_until = NULL
                 WHERE api_key_id = $1
                   AND idempotency_key = $2
                   AND request_hash = $3
+                  AND lease_owner = $6
                   AND response_status = 0
                 "#,
             )
@@ -211,16 +389,19 @@ pub async fn complete(
             .bind(request_hash)
             .bind(response_status)
             .bind(response_body)
-            .execute(&state.db)
+            .bind(owner)
+            .execute(&mut **tx)
             .await
         }
     }
-    .map_err(|err| AppError::new(ErrorCode::Internal, "写入幂等结果失败").with_source(err))?;
+    .map_err(|err| AppError::new(ErrorCode::Internal, "事务内写入幂等结果失败").with_source(err))?;
 
-    if updated.rows_affected() == 0 {
+    if updated.rows_affected() != 1 {
-        tracing::warn!("idempotency record not updated (maybe already completed?)");
+        return Err(AppError::new(
+            ErrorCode::IdempotencyConflict,
+            "幂等请求所有权已变化，请重试",
+        ));
     }
-
     Ok(())
 }
 
@@ -229,25 +410,28 @@ pub async fn abort(
     scope: Scope,
     idempotency_key: &str,
     request_hash: &str,
+    owner: Uuid,
 ) -> Result<(), AppError> {
     match scope {
         Scope::User(user_id) => {
             let _ = sqlx::query(
-                "DELETE FROM idempotency_keys WHERE user_id = $1 AND idempotency_key = $2 AND request_hash = $3 AND response_status = 0",
+                "DELETE FROM idempotency_keys WHERE user_id = $1 AND idempotency_key = $2 AND request_hash = $3 AND lease_owner = $4 AND response_status = 0",
             )
             .bind(user_id)
             .bind(idempotency_key)
             .bind(request_hash)
+            .bind(owner)
             .execute(&state.db)
             .await;
         }
         Scope::ApiKey(api_key_id) => {
             let _ = sqlx::query(
-                "DELETE FROM idempotency_keys WHERE api_key_id = $1 AND idempotency_key = $2 AND request_hash = $3 AND response_status = 0",
+                "DELETE FROM idempotency_keys WHERE api_key_id = $1 AND idempotency_key = $2 AND request_hash = $3 AND lease_owner = $4 AND response_status = 0",
             )
             .bind(api_key_id)
             .bind(idempotency_key)
             .bind(request_hash)
+            .bind(owner)
             .execute(&state.db)
             .await;
         }
@@ -334,3 +518,149 @@ async fn get_row(
 
     Ok(row)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::config::Config;
+    use crate::services::mail::Mailer;
+    use crate::services::settings::RuntimePolicyCache;
+    use crate::services::storage::StorageCache;
+    use sqlx::postgres::PgPoolOptions;
+    use std::sync::Arc;
+    use tokio::sync::Semaphore;
+
+    async fn test_state(database_url: String, redis_url: String) -> AppState {
+        let mut config = Config::from_env().expect("load idempotency test config");
+        config.database_url = database_url.clone();
+        config.redis_url = redis_url;
+        config.mail_enabled = false;
+        config.mail_log_links_when_disabled = false;
+        let db = PgPoolOptions::new()
+            .max_connections(8)
+            .connect(&database_url)
+            .await
+            .expect("connect idempotency test database");
+        sqlx::migrate!().run(&db).await.expect("run migrations");
+        let redis = redis::Client::open(config.redis_url.clone())
+            .expect("create idempotency test Redis client")
+            .get_connection_manager()
+            .await
+            .expect("connect idempotency test Redis");
+        AppState {
+            mailer: Arc::new(Mailer::new(&config).expect("create disabled test mailer")),
+            image_processing_semaphore: Arc::new(Semaphore::new(1)),
+            zip_build_semaphore: Arc::new(Semaphore::new(1)),
+            runtime_policy_cache: RuntimePolicyCache::new(),
+            storage_cache: StorageCache::new(),
+            config,
+            db,
+            redis,
+        }
+    }
+
+    #[tokio::test]
+    #[ignore = "requires IMAGEFORGE_TEST_DATABASE_URL and IMAGEFORGE_TEST_REDIS_URL"]
+    async fn stale_operation_is_fenced_and_replay_remains_atomic() {
+        let database_url = std::env::var("IMAGEFORGE_TEST_DATABASE_URL")
+            .expect("IMAGEFORGE_TEST_DATABASE_URL is required");
+        let redis_url = std::env::var("IMAGEFORGE_TEST_REDIS_URL")
+            .expect("IMAGEFORGE_TEST_REDIS_URL is required");
+        let state = test_state(database_url, redis_url).await;
+        let marker = Uuid::new_v4().simple().to_string();
+        let user_id: Uuid = sqlx::query_scalar(
+            r#"
+            INSERT INTO users (email, username, password_hash, email_verified_at)
+            VALUES ($1, $2, 'test', NOW())
+            RETURNING id
+            "#,
+        )
+        .bind(format!("idem-{marker}@example.test"))
+        .bind(format!("idem-{marker}"))
+        .fetch_one(&state.db)
+        .await
+        .expect("insert idempotency test user");
+        let scope = Scope::User(user_id);
+        let key = format!("idem-{marker}");
+        let request_hash = "a".repeat(64);
+        let owner_one = match begin(&state, scope, &key, &request_hash, 24)
+            .await
+            .expect("acquire first operation")
+        {
+            BeginResult::Acquired { owner } => owner,
+            other => panic!("unexpected first begin result: {other:?}"),
+        };
+        assert!(matches!(
+            begin(&state, scope, &key, &request_hash, 24)
+                .await
+                .expect("probe live operation"),
+            BeginResult::InProgress
+        ));
+
+        sqlx::query(
+            "UPDATE idempotency_keys SET lease_until = NOW() - INTERVAL '1 second' WHERE user_id = $1 AND idempotency_key = $2",
+        )
+        .bind(user_id)
+        .bind(&key)
+        .execute(&state.db)
+        .await
+        .expect("expire first operation lease");
+        let owner_two = match begin(&state, scope, &key, &request_hash, 24)
+            .await
+            .expect("take over stale operation")
+        {
+            BeginResult::Acquired { owner } => owner,
+            other => panic!("unexpected takeover result: {other:?}"),
+        };
+        assert_ne!(owner_one, owner_two);
+
+        let mut stale_tx = state.db.begin().await.expect("begin stale completion tx");
+        let stale_error = complete_in_tx(
+            &mut stale_tx,
+            scope,
+            &key,
+            &request_hash,
+            owner_one,
+            200,
+            serde_json::json!({"owner": "stale"}),
+        )
+        .await
+        .expect_err("stale operation completed after takeover");
+        assert_eq!(stale_error.code, ErrorCode::IdempotencyConflict);
+        stale_tx
+            .rollback()
+            .await
+            .expect("rollback stale completion");
+
+        let expected = serde_json::json!({"owner": "current"});
+        let mut current_tx = state.db.begin().await.expect("begin current completion tx");
+        complete_in_tx(
+            &mut current_tx,
+            scope,
+            &key,
+            &request_hash,
+            owner_two,
+            200,
+            expected.clone(),
+        )
+        .await
+        .expect("complete current operation");
+        current_tx
+            .commit()
+            .await
+            .expect("commit current completion");
+        match begin(&state, scope, &key, &request_hash, 24)
+            .await
+            .expect("replay completed operation")
+        {
+            BeginResult::Replay { response_body } => assert_eq!(response_body, expected),
+            other => panic!("unexpected replay result: {other:?}"),
+        }
+
+        sqlx::query("DELETE FROM users WHERE id = $1")
+            .bind(user_id)
+            .execute(&state.db)
+            .await
+            .expect("clean idempotency test user");
+    }
+}

src/services/mod.rs

@@ -6,7 +6,9 @@ pub mod filename;
 pub mod idempotency;
 pub mod mail;
 pub mod metrics;
+pub mod object_lifecycle;
 pub mod quota;
 pub mod rate_limit;
 pub mod settings;
 pub mod storage;
+pub mod task_queue;

src/services/storage.rs

@@ -330,27 +330,7 @@ fn retention_prefix(hours: i64) -> String {
     }
 }
 
-pub async fn store_bytes<B>(
+pub(crate) async fn store_bytes_s3(
-    state: &AppState,
-    key: &str,
-    bytes: B,
-    content_type: &str,
-) -> Result<StoredObject, AppError>
-where
-    B: Into<Bytes>,
-{
-    let bytes = bytes.into();
-    if let Some(endpoint) = active_endpoint(state).await? {
-        match store_bytes_s3(state, &endpoint, key, bytes.clone(), content_type).await {
-            Ok(stored) => return Ok(stored),
-            Err(err) => log_local_fallback(state, &endpoint, key, &err),
-        }
-    }
-
-    store_bytes_local(state, key, bytes.as_ref()).await
-}
-
-async fn store_bytes_s3(
     state: &AppState,
     endpoint: &StorageEndpoint,
     key: &str,
@@ -378,7 +358,7 @@ async fn store_bytes_s3(
     })
 }
 
-async fn store_bytes_local(
+pub(crate) async fn store_bytes_local(
     state: &AppState,
     key: &str,
     bytes: &[u8],
@@ -402,27 +382,7 @@ async fn store_bytes_local(
     })
 }
 
-pub async fn store_file(
+pub(crate) async fn store_file_s3(
-    state: &AppState,
-    key: &str,
-    path: &Path,
-    content_type: &str,
-) -> Result<StoredObject, AppError> {
-    let metadata = tokio::fs::metadata(path).await.map_err(|err| {
-        AppError::new(ErrorCode::StorageUnavailable, "读取待上传文件失败").with_source(err)
-    })?;
-
-    if let Some(endpoint) = active_endpoint(state).await? {
-        match store_file_s3(state, &endpoint, key, path, content_type, metadata.len()).await {
-            Ok(stored) => return Ok(stored),
-            Err(err) => log_local_fallback(state, &endpoint, key, &err),
-        }
-    }
-
-    store_file_local(state, key, path, metadata.len()).await
-}
-
-async fn store_file_s3(
     state: &AppState,
     endpoint: &StorageEndpoint,
     key: &str,
@@ -459,7 +419,7 @@ async fn store_file_s3(
     })
 }
 
-async fn store_file_local(
+pub(crate) async fn store_file_local(
     state: &AppState,
     key: &str,
     path: &Path,
@@ -484,7 +444,12 @@ async fn store_file_local(
     })
 }
 
-fn log_local_fallback(state: &AppState, endpoint: &StorageEndpoint, key: &str, err: &AppError) {
+pub(crate) fn log_local_fallback(
+    state: &AppState,
+    endpoint: &StorageEndpoint,
+    key: &str,
+    err: &AppError,
+) {
     crate::services::metrics::record_storage_fallback(state);
     tracing::warn!(
         storage_endpoint_id = %endpoint.id,
@@ -816,7 +781,7 @@ async fn endpoint_for_object(
     get_endpoint(state, endpoint_id).await
 }
 
-fn local_path(state: &AppState, key: &str) -> Result<PathBuf, AppError> {
+pub(crate) fn local_path(state: &AppState, key: &str) -> Result<PathBuf, AppError> {
     if key.is_empty()
         || key.starts_with('/')
         || key.starts_with('\\')

src/services/task_queue.rs

@@ -0,0 +1,367 @@
+use crate::error::{AppError, ErrorCode};
+use crate::services::{metrics, object_lifecycle, quota};
+use crate::state::AppState;
+
+use chrono::Utc;
+use sqlx::FromRow;
+use std::time::Duration;
+use uuid::Uuid;
+
+const DISPATCH_INTERVAL: Duration = Duration::from_secs(1);
+const DISPATCH_BATCH_SIZE: i64 = 50;
+const DELIVERY_LEASE_SECONDS: i64 = 30;
+const MAX_DELIVERY_ATTEMPTS: i32 = 20;
+const MAX_RETRY_SECONDS: u64 = 60;
+
+#[derive(Debug, FromRow)]
+struct OutboxClaim {
+    task_id: Uuid,
+    attempts: i32,
+}
+
+pub async fn dispatch_loop(state: AppState) {
+    let mut interval = tokio::time::interval(DISPATCH_INTERVAL);
+    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+    loop {
+        interval.tick().await;
+        if let Err(err) = dispatch_ready(&state, DISPATCH_BATCH_SIZE).await {
+            tracing::error!(error = %err, "task queue outbox dispatch iteration failed");
+        }
+    }
+}
+
+pub async fn dispatch_ready(state: &AppState, limit: i64) -> Result<usize, AppError> {
+    reconcile_non_pending_tasks(state).await?;
+    let lease_owner = Uuid::new_v4();
+    let claims = claim_ready(state, lease_owner, limit.max(1), None).await?;
+    let count = claims.len();
+    for claim in claims {
+        dispatch_claim(state, lease_owner, claim).await?;
+    }
+    Ok(count)
+}
+
+pub async fn dispatch_task(state: &AppState, task_id: Uuid) -> Result<bool, AppError> {
+    let lease_owner = Uuid::new_v4();
+    let mut claims = claim_ready(state, lease_owner, 1, Some(task_id)).await?;
+    let Some(claim) = claims.pop() else {
+        return Ok(false);
+    };
+    dispatch_claim(state, lease_owner, claim).await?;
+    Ok(true)
+}
+
+async fn claim_ready(
+    state: &AppState,
+    lease_owner: Uuid,
+    limit: i64,
+    task_id: Option<Uuid>,
+) -> Result<Vec<OutboxClaim>, AppError> {
+    sqlx::query_as::<_, OutboxClaim>(
+        r#"
+        WITH candidate AS (
+          SELECT outbox.task_id
+          FROM task_queue_outbox AS outbox
+          JOIN tasks AS task ON task.id = outbox.task_id
+          WHERE outbox.status IN ('pending', 'delivering')
+            AND outbox.next_attempt_at <= NOW()
+            AND (outbox.lease_until IS NULL OR outbox.lease_until <= NOW())
+            AND task.status = 'pending'
+            AND task.deletion_started_at IS NULL
+            AND ($3::uuid IS NULL OR outbox.task_id = $3)
+          ORDER BY outbox.next_attempt_at, outbox.created_at
+          FOR UPDATE OF outbox SKIP LOCKED
+          LIMIT $2
+        )
+        UPDATE task_queue_outbox AS outbox
+        SET status = 'delivering',
+            attempts = outbox.attempts + 1,
+            lease_owner = $1,
+            lease_until = NOW() + ($4 * INTERVAL '1 second'),
+            updated_at = NOW()
+        FROM candidate
+        WHERE outbox.task_id = candidate.task_id
+        RETURNING outbox.task_id, outbox.attempts
+        "#,
+    )
+    .bind(lease_owner)
+    .bind(limit)
+    .bind(task_id)
+    .bind(DELIVERY_LEASE_SECONDS)
+    .fetch_all(&state.db)
+    .await
+    .map_err(|err| AppError::new(ErrorCode::Internal, "领取任务队列 outbox 失败").with_source(err))
+}
+
+async fn dispatch_claim(
+    state: &AppState,
+    lease_owner: Uuid,
+    claim: OutboxClaim,
+) -> Result<(), AppError> {
+    match enqueue_task(state, claim.task_id).await {
+        Ok(()) => mark_delivered(state, claim.task_id, lease_owner).await,
+        Err(err) if claim.attempts >= MAX_DELIVERY_ATTEMPTS => {
+            dead_letter_pending_task(state, claim.task_id, lease_owner, &err).await
+        }
+        Err(err) => release_for_retry(state, claim, lease_owner, &err).await,
+    }
+}
+
+pub(crate) async fn enqueue_task(state: &AppState, task_id: Uuid) -> Result<(), AppError> {
+    let mut connection = state.redis.clone();
+    redis::cmd("XADD")
+        .arg(metrics::QUEUE_STREAM_KEY)
+        .arg("MAXLEN")
+        .arg("~")
+        .arg(100_000)
+        .arg("*")
+        .arg("task_id")
+        .arg(task_id.to_string())
+        .arg("created_at")
+        .arg(Utc::now().to_rfc3339())
+        .query_async::<_, redis::Value>(&mut connection)
+        .await
+        .map_err(|err| AppError::new(ErrorCode::Internal, "写入队列失败").with_source(err))?;
+    Ok(())
+}
+
+async fn mark_delivered(
+    state: &AppState,
+    task_id: Uuid,
+    lease_owner: Uuid,
+) -> Result<(), AppError> {
+    sqlx::query(
+        r#"
+        UPDATE task_queue_outbox
+        SET status = 'delivered',
+            delivered_at = COALESCE(delivered_at, NOW()),
+            lease_owner = NULL,
+            lease_until = NULL,
+            last_error = NULL,
+            updated_at = NOW()
+        WHERE task_id = $1
+          AND status = 'delivering'
+          AND lease_owner = $2
+        "#,
+    )
+    .bind(task_id)
+    .bind(lease_owner)
+    .execute(&state.db)
+    .await
+    .map_err(|err| {
+        AppError::new(ErrorCode::Internal, "完成任务队列 outbox 失败").with_source(err)
+    })?;
+    Ok(())
+}
+
+async fn release_for_retry(
+    state: &AppState,
+    claim: OutboxClaim,
+    lease_owner: Uuid,
+    error: &AppError,
+) -> Result<(), AppError> {
+    let delay = retry_delay(claim.attempts);
+    let message = truncate_error(error);
+    sqlx::query(
+        r#"
+        UPDATE task_queue_outbox
+        SET status = 'pending',
+            next_attempt_at = NOW() + ($3 * INTERVAL '1 second'),
+            lease_owner = NULL,
+            lease_until = NULL,
+            last_error = $4,
+            updated_at = NOW()
+        WHERE task_id = $1
+          AND status = 'delivering'
+          AND lease_owner = $2
+        "#,
+    )
+    .bind(claim.task_id)
+    .bind(lease_owner)
+    .bind(delay.as_secs() as i64)
+    .bind(&message)
+    .execute(&state.db)
+    .await
+    .map_err(|err| AppError::new(ErrorCode::Internal, "记录任务队列重试失败").with_source(err))?;
+    tracing::warn!(task_id = %claim.task_id, attempts = claim.attempts, retry_seconds = delay.as_secs(), error = %error, "task queue delivery deferred");
+    Ok(())
+}
+
+async fn dead_letter_pending_task(
+    state: &AppState,
+    task_id: Uuid,
+    lease_owner: Uuid,
+    error: &AppError,
+) -> Result<(), AppError> {
+    let message = format!("队列持续不可用：{}", truncate_error(error));
+    let input_dir = std::path::PathBuf::from(&state.config.storage_path)
+        .join("orig")
+        .join(task_id.to_string())
+        .to_string_lossy()
+        .to_string();
+    let mut tx = state.db.begin().await.map_err(|err| {
+        AppError::new(ErrorCode::Internal, "开启 outbox 死信事务失败").with_source(err)
+    })?;
+    let task: 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, "锁定 outbox 死信任务失败").with_source(err)
+            })?;
+    let Some(status) = task else {
+        tx.rollback().await.ok();
+        return Ok(());
+    };
+    let owned: Option<Uuid> = sqlx::query_scalar(
+        r#"
+        SELECT task_id
+        FROM task_queue_outbox
+        WHERE task_id = $1
+          AND status = 'delivering'
+          AND lease_owner = $2
+        FOR UPDATE
+        "#,
+    )
+    .bind(task_id)
+    .bind(lease_owner)
+    .fetch_optional(&mut *tx)
+    .await
+    .map_err(|err| {
+        AppError::new(ErrorCode::Internal, "校验 outbox 死信租约失败").with_source(err)
+    })?;
+    if owned.is_none() {
+        tx.rollback().await.ok();
+        return Ok(());
+    }
+
+    let failed = status == "pending";
+    if failed {
+        sqlx::query(
+            r#"
+            UPDATE task_files
+            SET status = 'failed',
+                error_message = $2,
+                completed_at = NOW(),
+                input_path = NULL,
+                lease_owner = NULL,
+                lease_until = NULL
+            WHERE task_id = $1
+              AND status IN ('pending', 'processing')
+            "#,
+        )
+        .bind(task_id)
+        .bind(&message)
+        .execute(&mut *tx)
+        .await
+        .map_err(|err| {
+            AppError::new(ErrorCode::Internal, "收口 outbox 死信文件失败").with_source(err)
+        })?;
+        sqlx::query(
+            r#"
+            INSERT INTO storage_objects (
+              task_id, object_kind, state, backend, object_key
+            ) VALUES ($1, 'input_dir', 'delete_pending', 'local_dir', $2)
+            ON CONFLICT DO NOTHING
+            "#,
+        )
+        .bind(task_id)
+        .bind(&input_dir)
+        .execute(&mut *tx)
+        .await
+        .map_err(|err| {
+            AppError::new(ErrorCode::Internal, "安排 outbox 死信输入清理失败").with_source(err)
+        })?;
+        sqlx::query(
+            r#"
+            UPDATE tasks
+            SET status = 'failed',
+                completed_files = 0,
+                failed_files = total_files,
+                error_message = $2,
+                completed_at = NOW(),
+                lease_owner = NULL,
+                lease_until = NULL
+            WHERE id = $1 AND status = 'pending'
+            "#,
+        )
+        .bind(task_id)
+        .bind(&message)
+        .execute(&mut *tx)
+        .await
+        .map_err(|err| {
+            AppError::new(ErrorCode::Internal, "收口 outbox 死信任务失败").with_source(err)
+        })?;
+    }
+
+    sqlx::query(
+        r#"
+        UPDATE task_queue_outbox
+        SET status = CASE WHEN $3 THEN 'dead' ELSE 'delivered' END,
+            delivered_at = CASE WHEN $3 THEN delivered_at ELSE COALESCE(delivered_at, NOW()) END,
+            lease_owner = NULL,
+            lease_until = NULL,
+            last_error = $4,
+            updated_at = NOW()
+        WHERE task_id = $1
+          AND lease_owner = $2
+        "#,
+    )
+    .bind(task_id)
+    .bind(lease_owner)
+    .bind(failed)
+    .bind(&message)
+    .execute(&mut *tx)
+    .await
+    .map_err(|err| {
+        AppError::new(ErrorCode::Internal, "提交 outbox 死信状态失败").with_source(err)
+    })?;
+    tx.commit().await.map_err(|err| {
+        AppError::new(ErrorCode::Internal, "提交 outbox 死信事务失败").with_source(err)
+    })?;
+
+    if failed {
+        metrics::record_dead_letter(state);
+        if let Err(err) = object_lifecycle::cleanup_ready_objects(state, 10, Some(task_id)).await {
+            tracing::warn!(task_id = %task_id, error = %err, "dead outbox input cleanup deferred");
+        }
+        quota::settle_anonymous_task_reservation(state, task_id).await?;
+    }
+    Ok(())
+}
+
+async fn reconcile_non_pending_tasks(state: &AppState) -> Result<(), AppError> {
+    sqlx::query(
+        r#"
+        UPDATE task_queue_outbox AS outbox
+        SET status = 'delivered',
+            delivered_at = COALESCE(outbox.delivered_at, NOW()),
+            lease_owner = NULL,
+            lease_until = NULL,
+            updated_at = NOW()
+        FROM tasks AS task
+        WHERE task.id = outbox.task_id
+          AND task.status <> 'pending'
+          AND outbox.status IN ('pending', 'delivering')
+        "#,
+    )
+    .execute(&state.db)
+    .await
+    .map_err(|err| {
+        AppError::new(ErrorCode::Internal, "对账任务队列 outbox 失败").with_source(err)
+    })?;
+    Ok(())
+}
+
+fn retry_delay(attempts: i32) -> Duration {
+    let exponent = attempts.saturating_sub(1).min(6) as u32;
+    Duration::from_secs(2_u64.saturating_pow(exponent).min(MAX_RETRY_SECONDS))
+}
+
+fn truncate_error(error: &AppError) -> String {
+    format!("{}: {}", error.code.as_str(), error.message)
+        .chars()
+        .take(2_000)
+        .collect()
+}

src/worker/mod.rs

@@ -2,6 +2,7 @@ use crate::error::{AppError, ErrorCode};
 use crate::services::billing;
 use crate::services::compress;
 use crate::services::metrics;
+use crate::services::object_lifecycle;
 use crate::services::quota;
 use crate::services::storage;
 use crate::state::AppState;
@@ -40,6 +41,10 @@ pub async fn run(state: AppState) -> Result<(), AppError> {
     let consumer = format!("worker_{worker_id}");
     ensure_group(&state).await?;
     tokio::spawn(maintenance_loop(state.clone()));
+    tokio::spawn(crate::services::task_queue::dispatch_loop(state.clone()));
+    tokio::spawn(crate::services::object_lifecycle::maintenance_loop(
+        state.clone(),
+    ));
 
     let task_concurrency = state.config.worker_task_concurrency.max(1) as usize;
     let mut inflight = JoinSet::new();
@@ -654,6 +659,7 @@ async fn ack_message(
 struct TaskProcRow {
     compression_level: String,
     compression_rate: Option<i16>,
+    target_size_bytes: Option<i64>,
     max_width: Option<i32>,
     max_height: Option<i32>,
     preserve_metadata: bool,
@@ -675,21 +681,6 @@ struct TaskFileProcRow {
     output_format: String,
 }
 
-#[derive(Debug, FromRow)]
-struct CleanupFileRow {
-    storage_backend: String,
-    storage_endpoint_id: Option<Uuid>,
-    storage_key: Option<String>,
-    input_path: Option<String>,
-}
-
-#[derive(Debug, FromRow)]
-struct CleanupZipRow {
-    zip_storage_backend: Option<String>,
-    zip_storage_endpoint_id: Option<Uuid>,
-    zip_storage_key: Option<String>,
-}
-
 #[derive(Clone)]
 struct TaskContext {
     api_key_id: Option<Uuid>,
@@ -703,12 +694,12 @@ struct TaskContext {
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
-enum TaskProcessOutcome {
+pub(crate) enum TaskProcessOutcome {
     Done,
     LeaseBusy,
 }
 
-async fn process_task(
+pub(crate) async fn process_task(
     state: &AppState,
     task_id: Uuid,
     worker_id: Uuid,
@@ -729,6 +720,7 @@ async fn process_task(
             lease_owner = $2,
             lease_until = NOW() + $3 * INTERVAL '1 second'
         WHERE id = $1
+          AND deletion_started_at IS NULL
           AND (
             status = 'pending'
             OR (
@@ -744,6 +736,7 @@ async fn process_task(
         RETURNING
           compression_level::text AS compression_level,
           compression_rate,
+          target_size_bytes,
           max_width,
           max_height,
           preserve_metadata,
@@ -791,6 +784,7 @@ async fn process_task(
     };
 
     let compression_rate = task.compression_rate.and_then(|v| u8::try_from(v).ok());
+    let target_size_bytes = task.target_size_bytes.and_then(|v| u64::try_from(v).ok());
     let level = compression_rate
         .map(compress::rate_to_level)
         .unwrap_or(compress::parse_level(&task.compression_level)?);
@@ -858,6 +852,7 @@ async fn process_task(
                 file,
                 level,
                 compression_rate,
+                target_size_bytes,
                 max_width,
                 max_height,
                 ctx,
@@ -940,6 +935,7 @@ async fn file_attempt_is_current(state: &AppState, fence: &FileFence) -> Result<
           JOIN task_files f ON f.task_id = t.id
           WHERE t.id = $1
             AND t.status = 'processing'
+            AND t.deletion_started_at IS NULL
             AND t.processing_attempt = $2
             AND t.lease_owner = $5
             AND t.lease_until > NOW()
@@ -961,25 +957,18 @@ async fn file_attempt_is_current(state: &AppState, fence: &FileFence) -> Result<
     .map_err(|err| AppError::new(ErrorCode::Internal, "检查文件处理租约失败").with_source(err))
 }
 
-#[allow(clippy::too_many_arguments)]
+async fn claim_task_file_attempt(
-async fn process_task_file(
+    state: &AppState,
-    state: AppState,
     task_id: Uuid,
     task_attempt: i64,
+    file_id: Uuid,
     worker_id: Uuid,
-    file: TaskFileProcRow,
+) -> Result<Option<i64>, AppError> {
-    level: compress::CompressionLevel,
+    sqlx::query_scalar(
-    compression_rate: Option<u8>,
-    max_width: Option<u32>,
-    max_height: Option<u32>,
-    ctx: TaskContext,
-    billing_ctx: Option<billing::BillingContext>,
-) -> Result<(), AppError> {
-    let file_attempt: Option<i64> = sqlx::query_scalar(
         r#"
         UPDATE task_files AS f
         SET status = 'processing',
-            processing_attempt = processing_attempt + 1,
+            processing_attempt = f.processing_attempt + 1,
             lease_owner = $4,
             lease_until = NOW() + $5 * INTERVAL '1 second',
             error_message = NULL
@@ -1006,14 +995,33 @@ async fn process_task_file(
         RETURNING f.processing_attempt
         "#,
     )
-    .bind(file.id)
+    .bind(file_id)
     .bind(task_id)
     .bind(task_attempt)
     .bind(worker_id)
     .bind(PROCESSING_LEASE_SECONDS)
     .fetch_optional(&state.db)
     .await
-    .map_err(|err| AppError::new(ErrorCode::Internal, "更新文件处理状态失败").with_source(err))?;
+    .map_err(|err| AppError::new(ErrorCode::Internal, "更新文件处理状态失败").with_source(err))
+}
+
+#[allow(clippy::too_many_arguments)]
+async fn process_task_file(
+    state: AppState,
+    task_id: Uuid,
+    task_attempt: i64,
+    worker_id: Uuid,
+    file: TaskFileProcRow,
+    level: compress::CompressionLevel,
+    compression_rate: Option<u8>,
+    target_size_bytes: Option<u64>,
+    max_width: Option<u32>,
+    max_height: Option<u32>,
+    ctx: TaskContext,
+    billing_ctx: Option<billing::BillingContext>,
+) -> Result<(), AppError> {
+    let file_attempt =
+        claim_task_file_attempt(&state, task_id, task_attempt, file.id, worker_id).await?;
     let Some(file_attempt) = file_attempt else {
         return Ok(());
     };
@@ -1061,7 +1069,7 @@ async fn process_task_file(
         format_out,
         level,
         compression_rate,
-        None, // target_size_bytes: worker 批量任务不支持精确大小
+        target_size_bytes,
         max_width,
         max_height,
         ctx.preserve_metadata,
@@ -1090,7 +1098,7 @@ async fn process_task_file(
         compression_rate,
         format_in == format_out,
         max_width.is_some() || max_height.is_some(),
-        false,
+        target_size_bytes.is_some(),
         original_size,
         compressed_size,
     );
@@ -1103,10 +1111,13 @@ async fn process_task_file(
         file_attempt,
         format_out.extension(),
     );
-    let stored = match storage::store_bytes(
+    let tracked = match object_lifecycle::store_tracked_bytes(
         &state,
+        task_id,
+        Some(file.id),
+        "result",
         &object_key,
-        compressed,
+        compressed.into(),
         format_out.content_type(),
     )
     .await
@@ -1122,19 +1133,19 @@ async fn process_task_file(
 
     if ctx.is_anonymous && charge_units && !ctx.anonymous_quota_reserved {
         let Some(session_id) = ctx.session_id.as_deref() else {
-            let _ = storage::delete_object(&state, &stored_locator(&stored)).await;
+            discard_tracked_result(&state, &tracked, None).await;
             mark_file_failed_and_cleanup(&state, &fence, "匿名任务缺少 session_id", &input_path)
                 .await?;
             return Ok(());
         };
         let Some(ip) = ctx.anon_ip else {
-            let _ = storage::delete_object(&state, &stored_locator(&stored)).await;
+            discard_tracked_result(&state, &tracked, None).await;
             mark_file_failed_and_cleanup(&state, &fence, "匿名任务缺少 client_ip", &input_path)
                 .await?;
             return Ok(());
         };
         if let Err(err) = quota::consume_anonymous_units(&state, session_id, ip, 1).await {
-            let _ = storage::delete_object(&state, &stored_locator(&stored)).await;
+            discard_tracked_result(&state, &tracked, Some(&err)).await;
             mark_file_failed_and_cleanup(&state, &fence, &err.message, &input_path).await?;
             return Ok(());
         }
@@ -1146,7 +1157,7 @@ async fn process_task_file(
         ctx.api_key_id,
         &ctx.source,
         &fence,
-        &stored,
+        &tracked,
         original_size as i64,
         compressed_size as i64,
         saved_percent,
@@ -1160,16 +1171,85 @@ async fn process_task_file(
             let _ = tokio::fs::remove_file(&input_path).await;
         }
         Ok(FinalizeFileOutcome::LeaseLost) => {
-            let _ = storage::delete_object(&state, &stored_locator(&stored)).await;
+            discard_tracked_result(&state, &tracked, None).await;
         }
-        Err(err) => {
+        Err(err) => match worker_result_was_committed(&state, &fence, &tracked).await {
-            let _ = storage::delete_object(&state, &stored_locator(&stored)).await;
+            Ok(true) => {
+                tracing::warn!(task_id = %task_id, file_id = %fence.file_id, error = %err, "worker result commit response was lost; recovered committed publication");
+                let _ = tokio::fs::remove_file(&input_path).await;
+            }
+            Ok(false) => {
+                discard_tracked_result(&state, &tracked, Some(&err)).await;
                 mark_file_failed_and_cleanup(&state, &fence, &err.message, &input_path).await?;
             }
+            Err(probe_err) => {
+                tracing::error!(task_id = %task_id, file_id = %fence.file_id, error = %probe_err, original_error = %err, "worker result commit state is unknown; staging lease will reconcile object");
+                return Err(err);
+            }
+        },
     }
     Ok(())
 }
 
+async fn discard_tracked_result(
+    state: &AppState,
+    tracked: &object_lifecycle::TrackedStoredObject,
+    error: Option<&AppError>,
+) {
+    if let Err(schedule_err) =
+        object_lifecycle::schedule_tracked_delete(state, tracked, error).await
+    {
+        tracing::error!(storage_object_id = %tracked.lifecycle_id, error = %schedule_err, "failed to persist discarded worker object cleanup");
+        return;
+    }
+    if let Err(cleanup_err) =
+        object_lifecycle::cleanup_ready_objects(state, 1, Some(tracked.task_id)).await
+    {
+        tracing::warn!(storage_object_id = %tracked.lifecycle_id, error = %cleanup_err, "discarded worker object cleanup deferred");
+    }
+}
+
+async fn worker_result_was_committed(
+    state: &AppState,
+    fence: &FileFence,
+    tracked: &object_lifecycle::TrackedStoredObject,
+) -> Result<bool, AppError> {
+    sqlx::query_scalar(
+        r#"
+        SELECT EXISTS(
+          SELECT 1
+          FROM tasks AS task
+          JOIN task_files AS file ON file.task_id = task.id
+          JOIN storage_objects AS object ON object.id = $3
+          WHERE task.id = $1
+            AND file.id = $2
+            AND file.status = 'completed'
+            AND file.storage_backend = $4
+            AND file.storage_endpoint_id IS NOT DISTINCT FROM $5
+            AND COALESCE(file.storage_key, file.storage_path) = $6
+            AND object.state = 'published'
+            AND object.task_id = task.id
+            AND object.task_file_id = file.id
+        )
+        "#,
+    )
+    .bind(fence.task_id)
+    .bind(fence.file_id)
+    .bind(tracked.lifecycle_id)
+    .bind(&tracked.stored.backend)
+    .bind(tracked.stored.endpoint_id)
+    .bind(&tracked.stored.key)
+    .fetch_one(&state.db)
+    .await
+    .map_err(|err| {
+        AppError::new(
+            ErrorCode::StorageUnavailable,
+            "核验 Worker 结果提交状态失败",
+        )
+        .with_source(err)
+    })
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 enum FinalizeFileOutcome {
     Committed,
@@ -1183,7 +1263,7 @@ async fn finalize_file(
     api_key_id: Option<Uuid>,
     source: &str,
     fence: &FileFence,
-    stored: &storage::StoredObject,
+    tracked: &object_lifecycle::TrackedStoredObject,
     bytes_in: i64,
     bytes_out: i64,
     saved_percent: f64,
@@ -1191,6 +1271,7 @@ async fn finalize_file(
     format_out: compress::ImageFmt,
     charge_units: bool,
 ) -> Result<FinalizeFileOutcome, AppError> {
+    let stored = &tracked.stored;
     let mut tx = state
         .db
         .begin()
@@ -1332,6 +1413,8 @@ async fn finalize_file(
         return Ok(FinalizeFileOutcome::LeaseLost);
     }
 
+    object_lifecycle::publish_in_tx(&mut tx, tracked).await?;
+
     tx.commit()
         .await
         .map_err(|err| AppError::new(ErrorCode::Internal, "提交事务失败").with_source(err))?;
@@ -1840,6 +1923,12 @@ async fn cleanup_expired_records(state: &AppState) -> Result<(), AppError> {
     .execute(&state.db)
     .await;
 
+    let _ = sqlx::query(
+        "DELETE FROM storage_objects WHERE state = 'deleted' AND deleted_at < NOW() - INTERVAL '7 days'",
+    )
+    .execute(&state.db)
+    .await;
+
     let _ =
         sqlx::query("DELETE FROM webhook_events WHERE received_at < NOW() - INTERVAL '90 days'")
             .execute(&state.db)
@@ -1851,6 +1940,7 @@ async fn cleanup_expired_records(state: &AppState) -> Result<(), AppError> {
         WHERE e.deleted_at < NOW() - INTERVAL '30 days'
           AND NOT EXISTS (SELECT 1 FROM task_files f WHERE f.storage_endpoint_id = e.id)
           AND NOT EXISTS (SELECT 1 FROM tasks t WHERE t.zip_storage_endpoint_id = e.id)
+          AND NOT EXISTS (SELECT 1 FROM storage_objects o WHERE o.storage_endpoint_id = e.id)
         "#,
     )
     .execute(&state.db)
@@ -1894,82 +1984,13 @@ async fn cleanup_expired_tasks(state: &AppState) -> Result<(), AppError> {
 }
 
 async fn cleanup_expired_task(state: &AppState, task_id: Uuid) -> Result<(), AppError> {
-    sqlx::query(
+    if object_lifecycle::mark_expired_task(state, task_id).await? {
-        "UPDATE tasks SET status = 'cancelled', completed_at = COALESCE(completed_at, NOW()) WHERE id = $1 AND expires_at < NOW() AND status IN ('pending', 'processing')",
+        object_lifecycle::finalize_task_deletion(state, task_id).await?;
-    )
-    .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,
-               COALESCE(storage_key, storage_path) AS storage_key,
-               input_path
-        FROM task_files
-        WHERE task_id = $1
-        "#,
-    )
-    .bind(task_id)
-    .fetch_all(&state.db)
-    .await
-    .map_err(|err| AppError::new(ErrorCode::Internal, "查询过期任务文件失败").with_source(err))?;
-
-    for file in files {
-        if let Some(key) = file.storage_key {
-            storage::delete_object(
-                state,
-                &storage::ObjectLocator {
-                    backend: file.storage_backend,
-                    endpoint_id: file.storage_endpoint_id,
-                    key,
-                },
-            )
-            .await?;
     }
-        if let Some(input_path) = file.input_path {
-            let _ = tokio::fs::remove_file(input_path).await;
-        }
-    }
-
-    let zip: Option<CleanupZipRow> = sqlx::query_as(
-        "SELECT zip_storage_backend, zip_storage_endpoint_id, zip_storage_key FROM tasks WHERE id = $1",
-    )
-    .bind(task_id)
-    .fetch_optional(&state.db)
-    .await
-    .map_err(|err| AppError::new(ErrorCode::Internal, "查询过期 ZIP 失败").with_source(err))?;
-    if let Some(zip) = zip {
-        if let (Some(backend), Some(key)) = (zip.zip_storage_backend, zip.zip_storage_key) {
-            storage::delete_object(
-                state,
-                &storage::ObjectLocator {
-                    backend,
-                    endpoint_id: zip.zip_storage_endpoint_id,
-                    key,
-                },
-            )
-            .await?;
-        }
-    }
-
-    let legacy_zip_path = format!("{}/zips/{task_id}.zip", state.config.storage_path);
-    let _ = tokio::fs::remove_file(legacy_zip_path).await;
-    let orig_dir = format!("{}/orig/{task_id}", state.config.storage_path);
-    let _ = tokio::fs::remove_dir_all(orig_dir).await;
-
-    sqlx::query("DELETE FROM tasks WHERE id = $1 AND expires_at < NOW()")
-        .bind(task_id)
-        .execute(&state.db)
-        .await
-        .map_err(|err| {
-            AppError::new(ErrorCode::Internal, "删除过期任务记录失败").with_source(err)
-        })?;
     Ok(())
 }
 
+#[cfg(test)]
 fn stored_locator(stored: &storage::StoredObject) -> storage::ObjectLocator {
     storage::ObjectLocator {
         backend: stored.backend.clone(),
@@ -1985,7 +2006,10 @@ mod tests {
     use crate::services::mail::Mailer;
     use bytes::Bytes;
     use chrono::Utc;
+    use image::{DynamicImage, ImageFormat, Rgb, RgbImage};
     use sqlx::postgres::PgPoolOptions;
+    use std::io::Cursor;
+    use std::path::PathBuf;
     use tokio::sync::Barrier;
 
     #[test]
@@ -2135,29 +2159,54 @@ mod tests {
               original_size, status, processing_attempt, lease_owner, lease_until
             ) VALUES (
               $1, $2, 'fence.png', 'png', 'png',
-              100, 'processing', 2, $3, NOW() + INTERVAL '5 minutes'
+              100, 'pending', 0, NULL, NULL
             )
             "#,
         )
         .bind(file_id)
         .bind(task_id)
-        .bind(winning_owner)
         .execute(&pool)
         .await
         .expect("insert test task file");
+        assert_eq!(
+            claim_task_file_attempt(&state, task_id, 2, file_id, winning_owner)
+                .await
+                .expect("claim pending task file"),
+            Some(1)
+        );
+        sqlx::query(
+            r#"
+            UPDATE task_files
+            SET processing_attempt = 2,
+                lease_owner = $2,
+                lease_until = NOW() + INTERVAL '5 minutes'
+            WHERE id = $1
+            "#,
+        )
+        .bind(file_id)
+        .bind(winning_owner)
+        .execute(&pool)
+        .await
+        .expect("prepare winning file fence");
 
         let stale_key = storage::result_attempt_key(24, task_id, file_id, 1, 1, "png");
         let winning_key = storage::result_attempt_key(24, task_id, file_id, 2, 2, "png");
-        let stale_object = storage::store_bytes(
+        let stale_object = object_lifecycle::store_tracked_bytes(
             &state,
+            task_id,
+            Some(file_id),
+            "result",
             &stale_key,
             Bytes::from_static(b"stale-attempt"),
             "image/png",
         )
         .await
         .expect("store stale attempt object");
-        let winning_object = storage::store_bytes(
+        let winning_object = object_lifecycle::store_tracked_bytes(
             &state,
+            task_id,
+            Some(file_id),
+            "result",
             &winning_key,
             Bytes::from_static(b"winning-attempt"),
             "image/png",
@@ -2165,8 +2214,8 @@ mod tests {
         .await
         .expect("store winning attempt object");
         if let Ok(expected_backend) = std::env::var("IMAGEFORGE_TEST_EXPECT_STORAGE_BACKEND") {
-            assert_eq!(stale_object.backend, expected_backend);
+            assert_eq!(stale_object.stored.backend, expected_backend);
-            assert_eq!(winning_object.backend, expected_backend);
+            assert_eq!(winning_object.stored.backend, expected_backend);
         }
 
         let period_start = Utc::now() - chrono::Duration::hours(1);
@@ -2259,14 +2308,14 @@ mod tests {
         assert_eq!(stale_result, FinalizeFileOutcome::LeaseLost);
         assert_eq!(winning_result, FinalizeFileOutcome::Committed);
 
-        storage::delete_object(&state, &stored_locator(&stale_object))
+        discard_tracked_result(&state, &stale_object, None).await;
+        assert!(
+            storage::read_bytes(&state, &stored_locator(&stale_object.stored))
                 .await
-            .expect("delete stale attempt object");
+                .is_err()
-        assert!(storage::read_bytes(&state, &stored_locator(&stale_object))
+        );
-            .await
-            .is_err());
         assert_eq!(
-            storage::read_bytes(&state, &stored_locator(&winning_object))
+            storage::read_bytes(&state, &stored_locator(&winning_object.stored))
                 .await
                 .expect("read winning object"),
             b"winning-attempt"
@@ -2292,7 +2341,11 @@ mod tests {
         .expect("query test file");
         assert_eq!(
             file,
-            ("completed".to_string(), winning_object.key.clone(), 40)
+            (
+                "completed".to_string(),
+                winning_object.stored.key.clone(),
+                40
+            )
         );
         let usage_event_count: i64 =
             sqlx::query_scalar("SELECT COUNT(*) FROM usage_events WHERE task_file_id = $1")
@@ -2312,9 +2365,124 @@ mod tests {
         .expect("query used units");
         assert_eq!(used_units, 1);
 
-        storage::delete_object(&state, &stored_locator(&winning_object))
+        let target_task_id = Uuid::new_v4();
+        let target_file_id = Uuid::new_v4();
+        let target_worker = Uuid::new_v4();
+        let target_image = DynamicImage::ImageRgb8(RgbImage::from_fn(160, 120, |x, y| {
+            let block = ((x / 20) + (y / 20) * 3) as u8;
+            Rgb([
+                block.wrapping_mul(31),
+                block.wrapping_mul(17),
+                block.wrapping_mul(11),
+            ])
+        }));
+        let mut input_cursor = Cursor::new(Vec::new());
+        target_image
+            .write_to(&mut input_cursor, ImageFormat::Png)
+            .expect("encode target-size input PNG");
+        let target_input = input_cursor.into_inner();
+        let target_input_dir = PathBuf::from(&state.config.storage_path)
+            .join("orig")
+            .join(target_task_id.to_string());
+        tokio::fs::create_dir_all(&target_input_dir)
             .await
-            .expect("delete winning object");
+            .expect("create target-size input directory");
+        let target_input_path = target_input_dir.join("source.png");
+        tokio::fs::write(&target_input_path, &target_input)
+            .await
+            .expect("write target-size input");
+
+        sqlx::query(
+            r#"
+            INSERT INTO tasks (
+              id, user_id, status, compression_level, output_format,
+              target_size_bytes, total_files, total_original_size,
+              expires_at, retention_hours
+            ) VALUES (
+              $1, $2, 'pending', 'medium', 'webp',
+              $3, 1, $4,
+              NOW() + INTERVAL '1 day', 24
+            )
+            "#,
+        )
+        .bind(target_task_id)
+        .bind(user_id)
+        .bind(1_048_576_i64)
+        .bind(target_input.len() as i64)
+        .execute(&pool)
+        .await
+        .expect("insert target-size task");
+        sqlx::query(
+            r#"
+            INSERT INTO task_files (
+              id, task_id, original_name, original_format, output_format,
+              original_size, input_path, status
+            ) VALUES ($1, $2, 'source.png', 'png', 'webp', $3, $4, 'pending')
+            "#,
+        )
+        .bind(target_file_id)
+        .bind(target_task_id)
+        .bind(target_input.len() as i64)
+        .bind(target_input_path.to_string_lossy().to_string())
+        .execute(&pool)
+        .await
+        .expect("insert target-size task file");
+
+        assert_eq!(
+            process_task(&state, target_task_id, target_worker)
+                .await
+                .expect("process target-size task"),
+            TaskProcessOutcome::Done
+        );
+        let target_task_status: String =
+            sqlx::query_scalar("SELECT status::text FROM tasks WHERE id = $1")
+                .bind(target_task_id)
+                .fetch_one(&pool)
+                .await
+                .expect("query target-size task status");
+        assert_eq!(target_task_status, "completed");
+        let target_result: (String, Option<Uuid>, String, i64) = sqlx::query_as(
+            r#"
+            SELECT storage_backend, storage_endpoint_id, storage_key, compressed_size
+            FROM task_files
+            WHERE id = $1
+            "#,
+        )
+        .bind(target_file_id)
+        .fetch_one(&pool)
+        .await
+        .expect("query target-size result");
+        assert!(target_result.3 <= 1_048_576);
+        let target_locator = storage::ObjectLocator {
+            backend: target_result.0,
+            endpoint_id: target_result.1,
+            key: target_result.2,
+        };
+        let target_output = storage::read_bytes(&state, &target_locator)
+            .await
+            .expect("read target-size result");
+        assert_eq!(
+            image::load_from_memory(&target_output)
+                .expect("decode target-size result")
+                .to_rgb8(),
+            target_image.to_rgb8(),
+            "worker must forward target_size_bytes and select the lossless candidate"
+        );
+        storage::delete_object(&state, &target_locator)
+            .await
+            .expect("delete target-size result object");
+        sqlx::query("DELETE FROM usage_events WHERE task_id = $1")
+            .bind(target_task_id)
+            .execute(&pool)
+            .await
+            .expect("delete target-size usage event");
+        sqlx::query("DELETE FROM tasks WHERE id = $1")
+            .bind(target_task_id)
+            .execute(&pool)
+            .await
+            .expect("delete target-size task");
+
+        discard_tracked_result(&state, &winning_object, None).await;
         sqlx::query("DELETE FROM usage_events WHERE task_id = $1")
             .bind(task_id)
             .execute(&pool)
@@ -2325,6 +2493,11 @@ mod tests {
             .execute(&pool)
             .await
             .expect("delete test task");
+        sqlx::query("DELETE FROM storage_objects WHERE task_id = $1")
+            .bind(task_id)
+            .execute(&pool)
+            .await
+            .expect("delete test storage lifecycle rows");
         sqlx::query("DELETE FROM usage_periods WHERE user_id = $1")
             .bind(user_id)
             .execute(&pool)