fix: make Stripe invoice webhooks monotonic

docs/billing.md

@@ -100,6 +100,7 @@ RETURNING used_units;
 说明：
 - 批量任务的计量仍以“成功文件数”为准；失败文件（含 `QUOTA_EXCEEDED`）不计费。
 - 前端建议在上传前调用 `GET /billing/usage`（登录）或读取配额头（API）做本地提示/拦截。
+- 匿名批量任务先按文件数预留当日额度，终态结算只退还失败或未完成文件。未提供 `compression_rate` 属于正常压缩并计量；只有显式 `compression_rate=100`、同格式且无缩放的原样请求免计量。
 
 ---
 
@@ -142,6 +143,7 @@ RETURNING used_units;
 - **乱序容忍**：订阅对象按 `(event.created, 事件优先级)` 保存独立水位；`deleted` 即使先到也会保留 tombstone，旧 `created/updated` 不得恢复已取消订阅。
 - **同秒歧义**：两个不同事件具有相同 `(event.created, 事件优先级)` 时，不能用不透明的 Event ID 排序，必须从 Stripe 拉取当前订阅快照并以快照响应时间推进水位。
 - **迁移对账**：历史版本用本地 `subscriptions.updated_at` 播种的非终态水位会标记为待对账；API 后台任务持租约获取 Stripe 快照，成功后才清除标记。未映射 Customer 或 Price 的受管订阅事件返回失败并等待重试，不能标记为已处理。
+- **发票一致性**：`invoices(provider, provider_invoice_id)` 唯一，发票事件也使用对象水位；新 `invoice.paid` 不会被迟到的旧 `invoice.payment_failed` 回退。同秒同等级事件从 Stripe 获取权威发票快照，未映射 Customer 时返回失败重试。
 - **并发一致性**：`subscriptions(provider, provider_subscription_id)` 唯一，订阅业务写入与 `webhook_events=processed` 在同一事务提交。
 - **可重放**：保存原始 payload（脱敏）用于排查。
 

docs/database.md

@@ -315,7 +315,7 @@ Stripe 运行时还通过迁移维护三组一致性结构：
 - `provider_object_event_watermarks` 以 Stripe `event.created` 和事件等级保存对象水位；同秒同等级的不同事件标记为歧义并触发权威快照，不能按 Event ID 字典序决定先后。
 - `stripe_subscription_reconciliations` 保存历史非因果水位的租约化对账任务，允许多 API 实例用 `FOR UPDATE SKIP LOCKED` 安全消费。
 
-数据库唯一索引同时保证非空 `users.billing_customer_id` 全局唯一、`subscriptions(provider, provider_subscription_id)` 唯一，以及每用户最多一条未取消 Stripe 订阅。部署这些索引前必须先清理存量冲突，具体检查见 `docs/deployment.md`。
+数据库唯一索引同时保证非空 `users.billing_customer_id` 全局唯一、`subscriptions(provider, provider_subscription_id)` 唯一、非空 `invoices(provider, provider_invoice_id)` 唯一，以及每用户最多一条未取消 Stripe 订阅。部署这些索引前必须先清理存量冲突，具体检查见 `docs/deployment.md`。
 
 ### 4.8 tasks - 压缩任务
 ```sql

docs/deployment.md

@@ -55,9 +55,9 @@ curl --fail http://127.0.0.1:8080/metrics
 
 ### 更新与回滚
 
-更新代码后保留 `.env.production` 和命名卷。包含迁移 `017`、`018` 的版本不能直接让旧、新 Worker 并行滚动：先备份数据库并停止旧 Worker，再构建新镜像。
+更新代码后保留 `.env.production` 和命名卷。包含迁移 `017` 至 `019` 的版本不能直接让旧、新 Worker 并行滚动：先备份数据库并停止旧 Worker，再构建新镜像。
 
-迁移 `017` 会在发现重复 Customer 或同用户多条未取消 Stripe 订阅时主动失败。部署前先检查并人工对账，两个查询都必须返回 0 行：
+迁移 `017` 会在发现重复 Customer 或同用户多条未取消 Stripe 订阅时主动失败，迁移 `019` 会在发现同一 Stripe 发票对应多行时主动失败。部署前先检查并人工对账，三个查询都必须返回 0 行：
 
 ```sql
 SELECT billing_customer_id, COUNT(*)
@@ -71,6 +71,12 @@ FROM subscriptions
 WHERE provider = 'stripe' AND status <> 'canceled'
 GROUP BY user_id
 HAVING COUNT(*) > 1;
+
+SELECT provider, provider_invoice_id, COUNT(*)
+FROM invoices
+WHERE provider_invoice_id IS NOT NULL
+GROUP BY provider, provider_invoice_id
+HAVING COUNT(*) > 1;
 ```
 
 推荐顺序：

migrations/019_invoice_webhook_invariants.sql

@@ -0,0 +1,16 @@
+DO $$
+BEGIN
+  IF EXISTS (
+    SELECT 1
+    FROM invoices
+    WHERE provider_invoice_id IS NOT NULL
+    GROUP BY provider, provider_invoice_id
+    HAVING COUNT(*) > 1
+  ) THEN
+    RAISE EXCEPTION 'duplicate invoices(provider, provider_invoice_id) values require reconciliation before migration 019';
+  END IF;
+END $$;
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_invoices_provider_object_unique
+  ON invoices(provider, provider_invoice_id)
+  WHERE provider_invoice_id IS NOT NULL;

src/api/compress.rs

@@ -299,12 +299,15 @@ async fn compress_json(
         } else {
             (saved_bytes as f64) * 100.0 / (original_size as f64)
         };
-        let skip_charge = req.compression_rate == Some(100)
-            && req.target_size_bytes.is_none()
-            && format_in == format_out
-            && req.max_width.is_none()
-            && req.max_height.is_none();
-        let charge_units = anonymous_reserved || (!skip_charge && compressed_size < original_size);
+        let charge_units = anonymous_reserved
+            || quota::output_consumes_unit(
+                req.compression_rate,
+                format_in == format_out,
+                req.max_width.is_some() || req.max_height.is_some(),
+                req.target_size_bytes.is_some(),
+                original_size,
+                compressed_size,
+            );
 
         let task_id = Uuid::new_v4();
         let file_id = Uuid::new_v4();
@@ -583,12 +586,14 @@ async fn compress_direct(
         } else {
             (saved_bytes as f64) * 100.0 / (original_size as f64)
         };
-        let skip_charge = req.compression_rate == Some(100)
-            && req.target_size_bytes.is_none()
-            && format_in == format_out
-            && req.max_width.is_none()
-            && req.max_height.is_none();
-        let charge_units = !skip_charge && compressed_size < original_size;
+        let charge_units = quota::output_consumes_unit(
+            req.compression_rate,
+            format_in == format_out,
+            req.max_width.is_some() || req.max_height.is_some(),
+            req.target_size_bytes.is_some(),
+            original_size,
+            compressed_size,
+        );
 
         let task_id = Uuid::new_v4();
         let file_id = Uuid::new_v4();

src/api/webhooks.rs

@@ -249,7 +249,9 @@ async fn process_stripe_event(
         "customer.subscription.deleted" => {
             apply_subscription_event(state, tx, event, &event.data.object).await
         }
-        "invoice.paid" | "invoice.payment_failed" => upsert_invoice(tx, &event.data.object).await,
+        "invoice.paid" | "invoice.payment_failed" => {
+            apply_invoice_event(state, tx, event, &event.data.object).await
+        }
         _ => Ok(()),
     }
 }
@@ -355,6 +357,13 @@ fn subscription_event_rank(event_type: &str) -> i16 {
     }
 }
 
+fn invoice_event_rank(event_type: &str) -> i16 {
+    match event_type {
+        "invoice.paid" => 1,
+        _ => 0,
+    }
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 enum WatermarkDecision {
     Apply,
@@ -362,27 +371,29 @@ enum WatermarkDecision {
     Reconcile,
 }
 
-async fn decide_subscription_event(
+async fn decide_object_event(
     tx: &mut Transaction<'_, Postgres>,
     event: &StripeEvent,
-    provider_subscription_id: &str,
+    object_type: &str,
+    provider_object_id: &str,
+    rank: i16,
     is_deleted: bool,
 ) -> Result<WatermarkDecision, AppError> {
-    let rank = subscription_event_rank(&event.type_);
     let inserted: Option<String> = sqlx::query_scalar(
         r#"
         INSERT INTO provider_object_event_watermarks (
           provider, object_type, provider_object_id,
           last_event_created, last_event_rank, last_event_id, is_deleted
         ) VALUES (
-          'stripe', 'subscription', $1,
-          $2, $3, $4, $5
+          'stripe', $1, $2,
+          $3, $4, $5, $6
         )
         ON CONFLICT (provider, object_type, provider_object_id) DO NOTHING
         RETURNING provider_object_id
         "#,
     )
-    .bind(provider_subscription_id)
+    .bind(object_type)
+    .bind(provider_object_id)
     .bind(event.created)
     .bind(rank)
     .bind(&event.id)
@@ -399,12 +410,13 @@ async fn decide_subscription_event(
         SELECT last_event_created, last_event_rank, last_event_id, requires_reconciliation
         FROM provider_object_event_watermarks
         WHERE provider = 'stripe'
-          AND object_type = 'subscription'
-          AND provider_object_id = $1
+          AND object_type = $1
+          AND provider_object_id = $2
         FOR UPDATE
         "#,
     )
-    .bind(provider_subscription_id)
+    .bind(object_type)
+    .bind(provider_object_id)
     .fetch_one(&mut **tx)
     .await
     .map_err(|err| AppError::new(ErrorCode::Internal, "锁定 Stripe 水位失败").with_source(err))?;
@@ -436,15 +448,16 @@ async fn decide_subscription_event(
             reconciliation_reason = NULL,
             updated_at = NOW()
         WHERE provider = 'stripe'
-          AND object_type = 'subscription'
+          AND object_type = $6
           AND provider_object_id = $1
         "#,
     )
-    .bind(provider_subscription_id)
+    .bind(provider_object_id)
     .bind(event.created)
     .bind(rank)
     .bind(&event.id)
     .bind(is_deleted)
+    .bind(object_type)
     .execute(&mut **tx)
     .await
     .map_err(|err| AppError::new(ErrorCode::Internal, "推进 Stripe 水位失败").with_source(err))?;
@@ -465,7 +478,7 @@ struct ResolvedSubscription {
     checkout_attempt_id: Option<uuid::Uuid>,
 }
 
-struct StripeSubscriptionSnapshot {
+struct StripeObjectSnapshot {
     object: serde_json::Value,
     reconciled_through: i64,
 }
@@ -481,17 +494,25 @@ async fn apply_subscription_event(
         .and_then(|value| value.as_str())
         .filter(|value| !value.trim().is_empty())
         .ok_or_else(|| AppError::new(ErrorCode::InvalidRequest, "subscription.id 缺失"))?;
-    let is_deleted = event.type_ == "customer.subscription.deleted"
-        || object.get("status").and_then(|value| value.as_str()) == Some("canceled");
-    match decide_subscription_event(tx, event, provider_subscription_id, is_deleted).await? {
+    let resolved = resolve_subscription(tx, object).await?;
+    let is_deleted =
+        event.type_ == "customer.subscription.deleted" || resolved.status == "canceled";
+    match decide_object_event(
+        tx,
+        event,
+        "subscription",
+        provider_subscription_id,
+        subscription_event_rank(&event.type_),
+        is_deleted,
+    )
+    .await?
+    {
         WatermarkDecision::Ignore => Ok(()),
-        WatermarkDecision::Apply => {
-            let resolved = resolve_subscription(tx, object).await?;
-            write_subscription(tx, &resolved).await
-        }
+        WatermarkDecision::Apply => write_subscription(tx, &resolved).await,
         WatermarkDecision::Reconcile => {
             let snapshot =
-                fetch_stripe_subscription_snapshot(state, provider_subscription_id).await?;
+                fetch_stripe_object_snapshot(state, "subscriptions", provider_subscription_id)
+                    .await?;
             let authoritative = resolve_subscription(tx, &snapshot.object).await?;
             if authoritative.provider_subscription_id != provider_subscription_id {
                 return Err(AppError::new(
@@ -500,7 +521,7 @@ async fn apply_subscription_event(
                 ));
             }
             write_subscription(tx, &authoritative).await?;
-            record_snapshot_watermark(
+            record_subscription_snapshot_watermark(
                 tx,
                 &authoritative,
                 snapshot.reconciled_through.max(event.created),
@@ -710,17 +731,18 @@ async fn write_subscription(
     Ok(())
 }
 
-async fn fetch_stripe_subscription_snapshot(
+async fn fetch_stripe_object_snapshot(
     state: &AppState,
-    provider_subscription_id: &str,
-) -> Result<StripeSubscriptionSnapshot, AppError> {
+    resource: &str,
+    provider_object_id: &str,
+) -> Result<StripeObjectSnapshot, AppError> {
     let secret = settings::get_stripe_secret(state)
         .await
         .map_err(|err| err.with_source("stripe secret not configured"))?;
-    let encoded_id = utf8_percent_encode(provider_subscription_id, NON_ALPHANUMERIC);
+    let encoded_id = utf8_percent_encode(provider_object_id, NON_ALPHANUMERIC);
     let url = format!(
-        "{}/v1/subscriptions/{encoded_id}",
-        state.config.stripe_api_base_url
+        "{}/v1/{resource}/{encoded_id}",
+        state.config.stripe_api_base_url,
     );
     let response = reqwest::Client::new()
         .get(url)
@@ -743,26 +765,28 @@ async fn fetch_stripe_subscription_snapshot(
         AppError::new(ErrorCode::Internal, "读取 Stripe 对账响应失败").with_source(err)
     })?;
     if !status.is_success() {
-        tracing::error!(%status, %body, provider_subscription_id, "Stripe subscription reconciliation failed");
+        tracing::error!(%status, %body, resource, provider_object_id, "Stripe object reconciliation failed");
         return Err(AppError::new(
             ErrorCode::StorageUnavailable,
-            "Stripe 订阅对账失败，将自动重试",
+            "Stripe 对象对账失败，将自动重试",
         ));
     }
     let object = serde_json::from_str(&body).map_err(|err| {
         AppError::new(ErrorCode::Internal, "解析 Stripe 对账响应失败").with_source(err)
     })?;
-    Ok(StripeSubscriptionSnapshot {
+    Ok(StripeObjectSnapshot {
         object,
         reconciled_through,
     })
 }
 
-async fn record_snapshot_watermark(
+async fn write_snapshot_watermark(
     tx: &mut Transaction<'_, Postgres>,
-    subscription: &ResolvedSubscription,
+    object_type: &str,
+    provider_object_id: &str,
     reconciled_through: i64,
     trigger_id: &str,
+    is_deleted: bool,
 ) -> Result<(), AppError> {
     sqlx::query(
         r#"
@@ -772,9 +796,9 @@ async fn record_snapshot_watermark(
           is_deleted, requires_reconciliation, reconciliation_reason,
           last_snapshot_at, updated_at
         ) VALUES (
-          'stripe', 'subscription', $1,
-          $2, 100, $3,
-          $4, false, NULL,
+          'stripe', $1, $2,
+          $3, 100, $4,
+          $5, false, NULL,
           NOW(), NOW()
         )
         ON CONFLICT (provider, object_type, provider_object_id) DO UPDATE
@@ -788,15 +812,34 @@ async fn record_snapshot_watermark(
             updated_at = NOW()
         "#,
     )
-    .bind(&subscription.provider_subscription_id)
+    .bind(object_type)
+    .bind(provider_object_id)
     .bind(reconciled_through)
     .bind(format!("snapshot:{trigger_id}"))
-    .bind(subscription.status == "canceled")
+    .bind(is_deleted)
     .execute(&mut **tx)
     .await
     .map_err(|err| {
         AppError::new(ErrorCode::Internal, "保存 Stripe 快照水位失败").with_source(err)
     })?;
+    Ok(())
+}
+
+async fn record_subscription_snapshot_watermark(
+    tx: &mut Transaction<'_, Postgres>,
+    subscription: &ResolvedSubscription,
+    reconciled_through: i64,
+    trigger_id: &str,
+) -> Result<(), AppError> {
+    write_snapshot_watermark(
+        tx,
+        "subscription",
+        &subscription.provider_subscription_id,
+        reconciled_through,
+        trigger_id,
+        subscription.status == "canceled",
+    )
+    .await?;
     sqlx::query(
         r#"
         UPDATE stripe_subscription_reconciliations
@@ -901,7 +944,8 @@ async fn reconcile_claimed_subscription(
     provider_subscription_id: &str,
     lease_owner: uuid::Uuid,
 ) -> Result<(), AppError> {
-    let snapshot = fetch_stripe_subscription_snapshot(state, provider_subscription_id).await?;
+    let snapshot =
+        fetch_stripe_object_snapshot(state, "subscriptions", provider_subscription_id).await?;
     let mut tx = state.db.begin().await.map_err(|err| {
         AppError::new(ErrorCode::Internal, "开启 Stripe 对账事务失败").with_source(err)
     })?;
@@ -956,7 +1000,7 @@ async fn reconcile_claimed_subscription(
     }
 
     write_subscription(&mut tx, &authoritative).await?;
-    record_snapshot_watermark(
+    record_subscription_snapshot_watermark(
         &mut tx,
         &authoritative,
         snapshot.reconciled_through,
@@ -969,133 +1013,211 @@ async fn reconcile_claimed_subscription(
     Ok(())
 }
 
-async fn upsert_invoice(
+#[derive(Debug)]
+struct ResolvedInvoice {
+    provider_invoice_id: String,
+    user_id: uuid::Uuid,
+    invoice_number: Option<String>,
+    fallback_invoice_number: String,
+    status: String,
+    currency: String,
+    total_amount_cents: i32,
+    period_start: Option<chrono::DateTime<Utc>>,
+    period_end: Option<chrono::DateTime<Utc>>,
+    hosted_invoice_url: Option<String>,
+    pdf_url: Option<String>,
+    paid_at: Option<chrono::DateTime<Utc>>,
+}
+
+async fn apply_invoice_event(
+    state: &AppState,
     tx: &mut Transaction<'_, Postgres>,
+    event: &StripeEvent,
     object: &serde_json::Value,
 ) -> Result<(), AppError> {
     let provider_invoice_id = object
         .get("id")
-        .and_then(|v| v.as_str())
+        .and_then(|value| value.as_str())
+        .filter(|value| !value.trim().is_empty())
+        .ok_or_else(|| AppError::new(ErrorCode::InvalidRequest, "invoice.id 缺失"))?;
+    let invoice = resolve_invoice(tx, object).await?;
+    match decide_object_event(
+        tx,
+        event,
+        "invoice",
+        provider_invoice_id,
+        invoice_event_rank(&event.type_),
+        false,
+    )
+    .await?
+    {
+        WatermarkDecision::Ignore => Ok(()),
+        WatermarkDecision::Apply => write_invoice(tx, &invoice).await,
+        WatermarkDecision::Reconcile => {
+            let snapshot =
+                fetch_stripe_object_snapshot(state, "invoices", provider_invoice_id).await?;
+            let authoritative = resolve_invoice(tx, &snapshot.object).await?;
+            if authoritative.provider_invoice_id != provider_invoice_id {
+                return Err(AppError::new(
+                    ErrorCode::Internal,
+                    "Stripe 对账快照发票 ID 不一致",
+                ));
+            }
+            write_invoice(tx, &authoritative).await?;
+            write_snapshot_watermark(
+                tx,
+                "invoice",
+                &authoritative.provider_invoice_id,
+                snapshot.reconciled_through.max(event.created),
+                &event.id,
+                false,
+            )
+            .await
+        }
+    }
+}
+
+async fn resolve_invoice(
+    tx: &mut Transaction<'_, Postgres>,
+    object: &serde_json::Value,
+) -> Result<ResolvedInvoice, AppError> {
+    let provider_invoice_id = object
+        .get("id")
+        .and_then(|value| value.as_str())
+        .filter(|value| !value.trim().is_empty())
         .ok_or_else(|| AppError::new(ErrorCode::InvalidRequest, "invoice.id 缺失"))?;
     let provider_customer_id = object
         .get("customer")
-        .and_then(|v| v.as_str())
+        .and_then(|value| value.as_str())
+        .filter(|value| !value.trim().is_empty())
         .ok_or_else(|| AppError::new(ErrorCode::InvalidRequest, "invoice.customer 缺失"))?;
 
     let user_id: Option<uuid::Uuid> =
-        sqlx::query_scalar("SELECT id FROM users WHERE billing_customer_id = $1 LIMIT 1")
+        sqlx::query_scalar("SELECT id FROM users WHERE billing_customer_id = $1 FOR UPDATE")
             .bind(provider_customer_id)
             .fetch_optional(&mut **tx)
             .await
-            .map_err(|err| AppError::new(ErrorCode::Internal, "查询用户失败").with_source(err))?;
-
-    let Some(user_id) = user_id else {
-        return Ok(());
-    };
-
-    let stripe_status = object
-        .get("status")
-        .and_then(|v| v.as_str())
-        .unwrap_or("open");
-    let status = map_invoice_status(stripe_status);
+            .map_err(|err| {
+                AppError::new(ErrorCode::Internal, "锁定发票用户失败").with_source(err)
+            })?;
+    let user_id = user_id.ok_or_else(|| {
+        AppError::new(
+            ErrorCode::StorageUnavailable,
+            "Stripe Customer 尚未映射到用户，发票事件将重试",
+        )
+    })?;
 
+    let status = map_invoice_status(
+        object
+            .get("status")
+            .and_then(|value| value.as_str())
+            .unwrap_or("open"),
+    )
+    .to_string();
     let invoice_number = object
         .get("number")
-        .and_then(|v| v.as_str())
-        .filter(|v| !v.trim().is_empty())
-        .map(|v| v.to_string())
-        .unwrap_or_else(|| format!("stripe_{provider_invoice_id}"));
-
+        .and_then(|value| value.as_str())
+        .filter(|value| !value.trim().is_empty())
+        .map(|value| truncate(value.to_string(), 50));
+    let fallback_invoice_number = truncate(format!("stripe_{provider_invoice_id}"), 50);
     let currency = object
         .get("currency")
-        .and_then(|v| v.as_str())
+        .and_then(|value| value.as_str())
         .unwrap_or("cny")
         .to_uppercase();
-    let total_amount_cents = object.get("total").and_then(|v| v.as_i64()).unwrap_or(0) as i32;
-
+    let total_amount_cents = i32::try_from(
+        object
+            .get("total")
+            .and_then(|value| value.as_i64())
+            .unwrap_or(0),
+    )
+    .map_err(|_| AppError::new(ErrorCode::InvalidRequest, "invoice.total 超出范围"))?;
     let hosted_invoice_url = object
         .get("hosted_invoice_url")
-        .and_then(|v| v.as_str())
-        .map(|v| v.to_string());
+        .and_then(|value| value.as_str())
+        .filter(|value| !value.trim().is_empty())
+        .map(str::to_string);
     let pdf_url = object
         .get("invoice_pdf")
-        .and_then(|v| v.as_str())
-        .map(|v| v.to_string());
-
+        .and_then(|value| value.as_str())
+        .filter(|value| !value.trim().is_empty())
+        .map(str::to_string);
     let period_start = object
         .get("period_start")
-        .and_then(|v| v.as_i64())
-        .and_then(|ts| Utc.timestamp_opt(ts, 0).single());
+        .and_then(|value| value.as_i64())
+        .and_then(|timestamp| Utc.timestamp_opt(timestamp, 0).single());
     let period_end = object
         .get("period_end")
-        .and_then(|v| v.as_i64())
-        .and_then(|ts| Utc.timestamp_opt(ts, 0).single());
-
+        .and_then(|value| value.as_i64())
+        .and_then(|timestamp| Utc.timestamp_opt(timestamp, 0).single());
     let paid_at = object
         .pointer("/status_transitions/paid_at")
-        .and_then(|v| v.as_i64())
-        .and_then(|ts| Utc.timestamp_opt(ts, 0).single());
+        .and_then(|value| value.as_i64())
+        .and_then(|timestamp| Utc.timestamp_opt(timestamp, 0).single());
 
-    let updated = sqlx::query(
+    Ok(ResolvedInvoice {
+        provider_invoice_id: provider_invoice_id.to_string(),
+        user_id,
+        invoice_number,
+        fallback_invoice_number,
+        status,
+        currency,
+        total_amount_cents,
+        period_start,
+        period_end,
+        hosted_invoice_url,
+        pdf_url,
+        paid_at,
+    })
+}
+
+async fn write_invoice(
+    tx: &mut Transaction<'_, Postgres>,
+    invoice: &ResolvedInvoice,
+) -> Result<(), AppError> {
+    sqlx::query(
         r#"
-        UPDATE invoices
-        SET status = $1::invoice_status,
-            currency = $2,
-            total_amount_cents = $3,
-            hosted_invoice_url = $4,
-            pdf_url = $5,
-            period_start = $6,
-            period_end = $7,
-            paid_at = $8
-        WHERE provider = 'stripe' AND provider_invoice_id = $9
+        INSERT INTO invoices (
+          user_id, invoice_number, status, currency, total_amount_cents,
+          period_start, period_end,
+          provider, provider_invoice_id, hosted_invoice_url, pdf_url,
+          paid_at
+        ) VALUES (
+          $1, COALESCE($2, $3), $4::invoice_status, $5, $6,
+          $7, $8,
+          'stripe', $9, $10, $11,
+          $12
+        )
+        ON CONFLICT (provider, provider_invoice_id)
+          WHERE provider_invoice_id IS NOT NULL
+        DO UPDATE SET
+          user_id = EXCLUDED.user_id,
+          invoice_number = COALESCE($2, invoices.invoice_number),
+          status = EXCLUDED.status,
+          currency = EXCLUDED.currency,
+          total_amount_cents = EXCLUDED.total_amount_cents,
+          hosted_invoice_url = COALESCE(EXCLUDED.hosted_invoice_url, invoices.hosted_invoice_url),
+          pdf_url = COALESCE(EXCLUDED.pdf_url, invoices.pdf_url),
+          period_start = COALESCE(EXCLUDED.period_start, invoices.period_start),
+          period_end = COALESCE(EXCLUDED.period_end, invoices.period_end),
+          paid_at = COALESCE(EXCLUDED.paid_at, invoices.paid_at)
         "#,
     )
-    .bind(status)
-    .bind(&currency)
-    .bind(total_amount_cents)
-    .bind(hosted_invoice_url.as_deref())
-    .bind(pdf_url.as_deref())
-    .bind(period_start)
-    .bind(period_end)
-    .bind(paid_at)
-    .bind(provider_invoice_id)
+    .bind(invoice.user_id)
+    .bind(invoice.invoice_number.as_deref())
+    .bind(&invoice.fallback_invoice_number)
+    .bind(&invoice.status)
+    .bind(&invoice.currency)
+    .bind(invoice.total_amount_cents)
+    .bind(invoice.period_start)
+    .bind(invoice.period_end)
+    .bind(&invoice.provider_invoice_id)
+    .bind(invoice.hosted_invoice_url.as_deref())
+    .bind(invoice.pdf_url.as_deref())
+    .bind(invoice.paid_at)
     .execute(&mut **tx)
     .await
-    .map_err(|err| AppError::new(ErrorCode::Internal, "更新发票失败").with_source(err))?;
-
-    if updated.rows_affected() == 0 {
-        let invoice_number = truncate(invoice_number, 50);
-        let _ = sqlx::query(
-            r#"
-            INSERT INTO invoices (
-              user_id, invoice_number, status, currency, total_amount_cents,
-              period_start, period_end,
-              provider, provider_invoice_id, hosted_invoice_url, pdf_url,
-              paid_at
-            ) VALUES (
-              $1, $2, $3::invoice_status, $4, $5,
-              $6, $7,
-              'stripe', $8, $9, $10,
-              $11
-            )
-            "#,
-        )
-        .bind(user_id)
-        .bind(invoice_number)
-        .bind(status)
-        .bind(&currency)
-        .bind(total_amount_cents)
-        .bind(period_start)
-        .bind(period_end)
-        .bind(provider_invoice_id)
-        .bind(hosted_invoice_url.as_deref())
-        .bind(pdf_url.as_deref())
-        .bind(paid_at)
-        .execute(&mut **tx)
-        .await
-        .map_err(|err| AppError::new(ErrorCode::Internal, "创建发票失败").with_source(err))?;
-    }
-
+    .map_err(|err| AppError::new(ErrorCode::Internal, "写入发票失败").with_source(err))?;
     Ok(())
 }
 
@@ -1153,10 +1275,10 @@ mod tests {
 
     async fn stripe_snapshot(
         State(mock): State<StripeSnapshotMock>,
-        Path(subscription_id): Path<String>,
+        Path(object_id): Path<String>,
     ) -> Response {
         mock.calls.fetch_add(1, Ordering::SeqCst);
-        let object = mock.objects.read().await.get(&subscription_id).cloned();
+        let object = mock.objects.read().await.get(&object_id).cloned();
         let mut headers = HeaderMap::new();
         headers.insert(
             header::DATE,
@@ -1181,6 +1303,7 @@ mod tests {
                 "/v1/subscriptions/{id}",
                 axum::routing::get(stripe_snapshot),
             )
+            .route("/v1/invoices/{id}", axum::routing::get(stripe_snapshot))
             .with_state(mock.clone());
         let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
             .await
@@ -1262,6 +1385,39 @@ mod tests {
         }
     }
 
+    fn invoice_event(
+        event_id: &str,
+        event_type: &str,
+        created: i64,
+        invoice_id: &str,
+        customer_id: &str,
+        invoice_number: Option<&str>,
+    ) -> StripeEvent {
+        let paid = event_type == "invoice.paid";
+        StripeEvent {
+            id: event_id.to_string(),
+            created,
+            type_: event_type.to_string(),
+            data: StripeEventData {
+                object: serde_json::json!({
+                    "id": invoice_id,
+                    "customer": customer_id,
+                    "number": invoice_number,
+                    "status": if paid { "paid" } else { "open" },
+                    "currency": "cny",
+                    "total": 1_900,
+                    "period_start": 1_700_000_000_i64,
+                    "period_end": 1_702_592_000_i64,
+                    "hosted_invoice_url": format!("https://billing.example.test/{invoice_id}"),
+                    "invoice_pdf": format!("https://billing.example.test/{invoice_id}.pdf"),
+                    "status_transitions": {
+                        "paid_at": if paid { Some(created) } else { None }
+                    }
+                }),
+            },
+        }
+    }
+
     async fn apply_test_event(state: &AppState, event: &StripeEvent) -> Result<(), AppError> {
         let mut tx = state.db.begin().await.expect("begin event transaction");
         match process_stripe_event(state, &mut tx, event).await {
@@ -1296,6 +1452,29 @@ mod tests {
         assert!(watermark.0);
     }
 
+    async fn assert_invoice_once(
+        pool: &sqlx::PgPool,
+        invoice_id: &str,
+        expected_status: &str,
+        expected_number: &str,
+    ) {
+        let rows: Vec<(String, String, Option<chrono::DateTime<Utc>>)> = sqlx::query_as(
+            r#"
+            SELECT status::text, invoice_number, paid_at
+            FROM invoices
+            WHERE provider = 'stripe' AND provider_invoice_id = $1
+            "#,
+        )
+        .bind(invoice_id)
+        .fetch_all(pool)
+        .await
+        .expect("query Stripe invoice");
+        assert_eq!(rows.len(), 1, "Stripe invoice was not unique");
+        assert_eq!(rows[0].0, expected_status);
+        assert_eq!(rows[0].1, expected_number);
+        assert_eq!(rows[0].2.is_some(), expected_status == "paid");
+    }
+
     #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
     #[ignore = "requires isolated IMAGEFORGE_TEST_DATABASE_URL and IMAGEFORGE_TEST_REDIS_URL"]
     async fn subscription_events_are_monotonic_for_all_orders_and_concurrency() {
@@ -1723,9 +1902,174 @@ mod tests {
             .await
             .expect("cancel primary subscription");
 
+        let concurrent_invoice_id = format!("inv_{marker}_concurrent");
+        let concurrent_invoice_number = format!("INV-{marker}-C");
+        let concurrent_invoice_events = [
+            invoice_event(
+                &format!("evt_{marker}_invoice_failed_concurrent"),
+                "invoice.payment_failed",
+                1_700_006_000,
+                &concurrent_invoice_id,
+                &customer_id,
+                Some(&concurrent_invoice_number),
+            ),
+            invoice_event(
+                &format!("evt_{marker}_invoice_paid_concurrent"),
+                "invoice.paid",
+                1_700_006_100,
+                &concurrent_invoice_id,
+                &customer_id,
+                Some(&concurrent_invoice_number),
+            ),
+        ];
+        let barrier = Arc::new(Barrier::new(concurrent_invoice_events.len()));
+        let mut joins = Vec::new();
+        for event in concurrent_invoice_events {
+            let state = state.clone();
+            let barrier = barrier.clone();
+            joins.push(tokio::spawn(async move {
+                barrier.wait().await;
+                apply_test_event(&state, &event).await
+            }));
+        }
+        for join in joins {
+            join.await
+                .expect("join concurrent invoice event")
+                .expect("apply concurrent invoice event");
+        }
+        assert_invoice_once(
+            &pool,
+            &concurrent_invoice_id,
+            "paid",
+            &concurrent_invoice_number,
+        )
+        .await;
+
+        let ordered_invoice_id = format!("inv_{marker}_ordered");
+        let ordered_invoice_number = format!("INV-{marker}-O");
+        let paid = invoice_event(
+            &format!("evt_{marker}_invoice_paid_new"),
+            "invoice.paid",
+            1_700_007_000,
+            &ordered_invoice_id,
+            &customer_id,
+            Some(&ordered_invoice_number),
+        );
+        let stale_failed = invoice_event(
+            &format!("evt_{marker}_invoice_failed_old"),
+            "invoice.payment_failed",
+            1_700_006_900,
+            &ordered_invoice_id,
+            &customer_id,
+            Some(&ordered_invoice_number),
+        );
+        apply_test_event(&state, &paid)
+            .await
+            .expect("apply newer paid invoice");
+        apply_test_event(&state, &stale_failed)
+            .await
+            .expect("ignore stale payment failure");
+        assert_invoice_once(&pool, &ordered_invoice_id, "paid", &ordered_invoice_number).await;
+
+        let numbered_invoice_id = format!("inv_{marker}_numbered");
+        let fallback_number = truncate(format!("stripe_{numbered_invoice_id}"), 50);
+        let actual_number = format!("INV-{marker}-N");
+        let missing_number = invoice_event(
+            &format!("evt_{marker}_invoice_missing_number"),
+            "invoice.payment_failed",
+            1_700_008_000,
+            &numbered_invoice_id,
+            &customer_id,
+            None,
+        );
+        let with_number = invoice_event(
+            &format!("evt_{marker}_invoice_with_number"),
+            "invoice.payment_failed",
+            1_700_008_100,
+            &numbered_invoice_id,
+            &customer_id,
+            Some(&actual_number),
+        );
+        apply_test_event(&state, &missing_number)
+            .await
+            .expect("apply invoice without number");
+        assert_invoice_once(&pool, &numbered_invoice_id, "open", &fallback_number).await;
+        apply_test_event(&state, &with_number)
+            .await
+            .expect("apply invoice number update");
+        assert_invoice_once(&pool, &numbered_invoice_id, "open", &actual_number).await;
+
+        let ambiguous_invoice_id = format!("inv_{marker}_ambiguous");
+        let ambiguous_number = format!("INV-{marker}-A");
+        let ambiguous_a = invoice_event(
+            &format!("evt_{marker}_invoice_ambiguous_a"),
+            "invoice.payment_failed",
+            1_700_009_000,
+            &ambiguous_invoice_id,
+            &customer_id,
+            Some(&ambiguous_number),
+        );
+        let ambiguous_z = invoice_event(
+            &format!("evt_{marker}_invoice_ambiguous_z"),
+            "invoice.payment_failed",
+            1_700_009_000,
+            &ambiguous_invoice_id,
+            &customer_id,
+            Some(&ambiguous_number),
+        );
+        let authoritative_invoice = invoice_event(
+            &format!("evt_{marker}_invoice_snapshot"),
+            "invoice.paid",
+            1_700_009_100,
+            &ambiguous_invoice_id,
+            &customer_id,
+            Some(&ambiguous_number),
+        );
+        stripe_mock.objects.write().await.insert(
+            ambiguous_invoice_id.clone(),
+            authoritative_invoice.data.object,
+        );
+        apply_test_event(&state, &ambiguous_a)
+            .await
+            .expect("apply first same-second invoice event");
+        apply_test_event(&state, &ambiguous_z)
+            .await
+            .expect("reconcile same-second invoice ambiguity");
+        assert_invoice_once(&pool, &ambiguous_invoice_id, "paid", &ambiguous_number).await;
+
+        let unmapped_invoice_id = format!("inv_{marker}_unmapped");
+        let unmapped_invoice = invoice_event(
+            &format!("evt_{marker}_invoice_unmapped"),
+            "invoice.paid",
+            1_700_010_000,
+            &unmapped_invoice_id,
+            &format!("cus_{marker}_invoice_unmapped"),
+            Some(&format!("INV-{marker}-U")),
+        );
+        let unmapped_invoice_error = apply_test_event(&state, &unmapped_invoice)
+            .await
+            .expect_err("unmapped invoice was marked processed");
+        assert_eq!(unmapped_invoice_error.code, ErrorCode::StorageUnavailable);
+        let unmapped_invoice_rows: i64 = sqlx::query_scalar(
+            "SELECT COUNT(*) FROM invoices WHERE provider = 'stripe' AND provider_invoice_id = $1",
+        )
+        .bind(&unmapped_invoice_id)
+        .fetch_one(&pool)
+        .await
+        .expect("count unmapped invoices");
+        assert_eq!(unmapped_invoice_rows, 0);
+        let unmapped_invoice_watermarks: i64 = sqlx::query_scalar(
+            "SELECT COUNT(*) FROM provider_object_event_watermarks WHERE object_type = 'invoice' AND provider_object_id = $1",
+        )
+        .bind(&unmapped_invoice_id)
+        .fetch_one(&pool)
+        .await
+        .expect("count unmapped invoice watermarks");
+        assert_eq!(unmapped_invoice_watermarks, 0);
+
         assert_eq!(
             stripe_mock.calls.load(Ordering::SeqCst),
-            4,
+            5,
             "unexpected number of authoritative Stripe snapshots"
         );
 
@@ -1743,6 +2087,13 @@ mod tests {
         .execute(&pool)
         .await
         .expect("delete test watermarks");
+        sqlx::query(
+            "DELETE FROM provider_object_event_watermarks WHERE object_type = 'invoice' AND provider_object_id LIKE $1",
+        )
+        .bind(format!("inv_{marker}%"))
+        .execute(&pool)
+        .await
+        .expect("delete test invoice watermarks");
         sqlx::query("DELETE FROM users WHERE id = $1")
             .bind(user_id)
             .execute(&pool)
@@ -1755,4 +2106,113 @@ mod tests {
             .expect("delete test plan");
         stripe_mock_task.abort();
     }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    #[ignore = "requires an isolated IMAGEFORGE_TEST_DATABASE_URL with CREATE DATABASE"]
+    async fn invoice_migration_rejects_ambiguous_duplicate_history() {
+        let database_url = std::env::var("IMAGEFORGE_TEST_DATABASE_URL")
+            .expect("IMAGEFORGE_TEST_DATABASE_URL must be set");
+        assert!(
+            database_url.to_ascii_lowercase().contains("test"),
+            "refusing to run destructive integration test outside a test database"
+        );
+        let mut admin_url = url::Url::parse(&database_url).expect("parse test database URL");
+        admin_url.set_path("/postgres");
+        let admin_pool = PgPoolOptions::new()
+            .max_connections(1)
+            .connect(admin_url.as_str())
+            .await
+            .expect("connect PostgreSQL admin database");
+        let marker = Uuid::new_v4().simple().to_string();
+        let child_database = format!("imageforge_test_invoice_{marker}");
+        sqlx::query(&format!("CREATE DATABASE {child_database}"))
+            .execute(&admin_pool)
+            .await
+            .expect("create invoice migration test database");
+
+        let mut child_url = url::Url::parse(&database_url).expect("parse child database URL");
+        child_url.set_path(&format!("/{child_database}"));
+        let child_pool = PgPoolOptions::new()
+            .max_connections(2)
+            .connect(child_url.as_str())
+            .await
+            .expect("connect invoice migration test database");
+        let all_migrations = sqlx::migrate!();
+        let legacy_migrator = sqlx::migrate::Migrator {
+            migrations: std::borrow::Cow::Owned(
+                all_migrations
+                    .iter()
+                    .filter(|migration| migration.version < 19)
+                    .cloned()
+                    .collect(),
+            ),
+            ignore_missing: false,
+            locking: true,
+            no_tx: false,
+        };
+        legacy_migrator
+            .run(&child_pool)
+            .await
+            .expect("run migrations through 018");
+
+        let first_user = Uuid::new_v4();
+        let second_user = Uuid::new_v4();
+        sqlx::query(
+            r#"
+            INSERT INTO users (id, email, username, password_hash)
+            VALUES
+              ($1, $2, $3, 'test-only'),
+              ($4, $5, $6, 'test-only')
+            "#,
+        )
+        .bind(first_user)
+        .bind(format!("invoice-migration-a-{marker}@example.test"))
+        .bind(format!("inv_a_{marker}"))
+        .bind(second_user)
+        .bind(format!("invoice-migration-b-{marker}@example.test"))
+        .bind(format!("inv_b_{marker}"))
+        .execute(&child_pool)
+        .await
+        .expect("insert duplicate-history users");
+        let duplicate_invoice_id = format!("in_duplicate_{marker}");
+        sqlx::query(
+            r#"
+            INSERT INTO invoices (
+              user_id, invoice_number, status, provider, provider_invoice_id
+            ) VALUES
+              ($1, $2, 'open', 'stripe', $3),
+              ($4, $5, 'paid', 'stripe', $3)
+            "#,
+        )
+        .bind(first_user)
+        .bind(format!("MIG-A-{marker}"))
+        .bind(&duplicate_invoice_id)
+        .bind(second_user)
+        .bind(format!("MIG-B-{marker}"))
+        .execute(&child_pool)
+        .await
+        .expect("insert duplicate invoice history");
+
+        let migration_result = all_migrations.run(&child_pool).await;
+        let index_exists: Option<String> =
+            sqlx::query_scalar("SELECT to_regclass('idx_invoices_provider_object_unique')::text")
+                .fetch_one(&child_pool)
+                .await
+                .expect("query invoice unique index");
+        child_pool.close().await;
+        sqlx::query(&format!("DROP DATABASE {child_database} WITH (FORCE)"))
+            .execute(&admin_pool)
+            .await
+            .expect("drop invoice migration test database");
+        admin_pool.close().await;
+
+        let migration_error = migration_result.expect_err("migration accepted duplicate invoices");
+        assert!(
+            migration_error
+                .to_string()
+                .contains("duplicate invoices(provider, provider_invoice_id)"),
+            "unexpected migration error: {migration_error}"
+        );
+        assert!(index_exists.is_none(), "unique index was partially applied");
+    }
 }

src/services/quota.rs

@@ -451,7 +451,8 @@ pub async fn settle_anonymous_task_reservation(
                    AND f.status = 'completed'
                    AND f.compressed_size < f.original_size
                    AND NOT (
-                     tasks.compression_rate = 100
+                     -- NULL means the caller did not request the explicit 100% passthrough.
+                     COALESCE(tasks.compression_rate = 100, false)
                      AND f.original_format = f.output_format
                      AND tasks.max_width IS NULL
                      AND tasks.max_height IS NULL
@@ -514,6 +515,19 @@ pub async fn settle_anonymous_task_reservation(
     Ok(Some(refundable))
 }
 
+pub(crate) fn output_consumes_unit(
+    compression_rate: Option<u8>,
+    same_format: bool,
+    has_resize: bool,
+    has_target_size: bool,
+    original_size: u64,
+    output_size: u64,
+) -> bool {
+    let is_unmetered_passthrough =
+        compression_rate == Some(100) && same_format && !has_resize && !has_target_size;
+    !is_unmetered_passthrough && output_size < original_size
+}
+
 fn refundable_reserved_units(reserved: i32, total_files: i32, consumed_units: i32) -> u32 {
     let reserved = reserved.max(0);
     let total_files = total_files.max(0);
@@ -556,6 +570,125 @@ fn utc8_date() -> NaiveDate {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::config::Config;
+    use crate::services::mail::Mailer;
+    use sqlx::postgres::PgPoolOptions;
+    use std::sync::Arc;
+    use tokio::sync::Semaphore;
+
+    struct AnonymousSettlementFixture<'a> {
+        session_id: &'a str,
+        ip: IpAddr,
+        date: NaiveDate,
+        reserved_units: u32,
+        compression_rate: Option<i16>,
+        total_files: usize,
+        completed_files: usize,
+    }
+
+    async fn build_test_state(
+        pool: sqlx::PgPool,
+        database_url: String,
+        redis_url: String,
+    ) -> AppState {
+        let mut config = Config::from_env().expect("load quota test config");
+        config.database_url = database_url;
+        config.redis_url = redis_url;
+        config.mail_enabled = false;
+        config.mail_log_links_when_disabled = false;
+        config.anon_daily_units = 10;
+        let redis = redis::Client::open(config.redis_url.clone())
+            .expect("create quota test Redis client")
+            .get_connection_manager()
+            .await
+            .expect("connect quota test Redis");
+        AppState {
+            mailer: Arc::new(Mailer::new(&config).expect("create disabled quota test mailer")),
+            image_processing_semaphore: Arc::new(Semaphore::new(2)),
+            runtime_policy_cache: crate::services::settings::RuntimePolicyCache::new(),
+            storage_cache: crate::services::storage::StorageCache::new(),
+            config,
+            db: pool,
+            redis,
+        }
+    }
+
+    async fn insert_anonymous_settlement_fixture(
+        pool: &sqlx::PgPool,
+        fixture: AnonymousSettlementFixture<'_>,
+    ) -> Uuid {
+        assert!(fixture.completed_files <= fixture.total_files);
+        let task_id = Uuid::new_v4();
+        sqlx::query(
+            r#"
+            INSERT INTO tasks (
+              id, session_id, client_ip, status,
+              compression_rate, total_files, completed_files, failed_files,
+              anonymous_units_reserved, anonymous_quota_date
+            ) VALUES (
+              $1, $2, $3::inet, 'completed',
+              $4, $5, $6, $7,
+              $8, $9
+            )
+            "#,
+        )
+        .bind(task_id)
+        .bind(fixture.session_id)
+        .bind(fixture.ip.to_string())
+        .bind(fixture.compression_rate)
+        .bind(fixture.total_files as i32)
+        .bind(fixture.completed_files as i32)
+        .bind((fixture.total_files - fixture.completed_files) as i32)
+        .bind(fixture.reserved_units as i32)
+        .bind(fixture.date)
+        .execute(pool)
+        .await
+        .expect("insert anonymous settlement task");
+
+        for index in 0..fixture.total_files {
+            let completed = index < fixture.completed_files;
+            sqlx::query(
+                r#"
+                INSERT INTO task_files (
+                  id, task_id, original_name, original_format, output_format,
+                  original_size, compressed_size, status
+                ) VALUES (
+                  $1, $2, $3, 'jpeg', 'jpeg',
+                  100, $4, $5::file_status
+                )
+                "#,
+            )
+            .bind(Uuid::new_v4())
+            .bind(task_id)
+            .bind(format!("fixture-{index}.jpg"))
+            .bind(completed.then_some(50_i64))
+            .bind(if completed { "completed" } else { "failed" })
+            .execute(pool)
+            .await
+            .expect("insert anonymous settlement file");
+        }
+        task_id
+    }
+
+    async fn anonymous_quota_counts(
+        state: &AppState,
+        session_id: &str,
+        ip: IpAddr,
+        date: NaiveDate,
+    ) -> (i64, i64) {
+        let mut redis = state.redis.clone();
+        let session_count: Option<i64> = redis::cmd("GET")
+            .arg(anonymous_session_key(session_id, date))
+            .query_async(&mut redis)
+            .await
+            .expect("read anonymous session quota");
+        let ip_count: Option<i64> = redis::cmd("GET")
+            .arg(anonymous_ip_key(ip, date))
+            .query_async(&mut redis)
+            .await
+            .expect("read anonymous IP quota");
+        (session_count.unwrap_or(0), ip_count.unwrap_or(0))
+    }
 
     #[test]
     fn balance_keeps_redeemed_units_separate_from_plan_usage() {
@@ -606,6 +739,29 @@ mod tests {
         assert_eq!(refundable_reserved_units(10, -1, -2), 0);
     }
 
+    #[test]
+    fn output_metering_matches_passthrough_contract() {
+        assert!(output_consumes_unit(None, true, false, false, 100, 50));
+        assert!(!output_consumes_unit(
+            Some(100),
+            true,
+            false,
+            false,
+            100,
+            50
+        ));
+        assert!(output_consumes_unit(
+            Some(100),
+            false,
+            false,
+            false,
+            100,
+            50
+        ));
+        assert!(output_consumes_unit(Some(100), true, true, false, 100, 50));
+        assert!(!output_consumes_unit(None, true, false, false, 100, 100));
+    }
+
     #[test]
     fn anonymous_quota_keys_use_the_reserved_date() {
         let date = NaiveDate::from_ymd_opt(2026, 7, 25).unwrap();
@@ -633,4 +789,194 @@ mod tests {
             "203.0.113.7"
         );
     }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    #[ignore = "requires isolated IMAGEFORGE_TEST_DATABASE_URL and IMAGEFORGE_TEST_REDIS_URL"]
+    async fn anonymous_batch_settlement_charges_successes_and_refunds_only_unused_units() {
+        let database_url = std::env::var("IMAGEFORGE_TEST_DATABASE_URL")
+            .expect("IMAGEFORGE_TEST_DATABASE_URL must be set");
+        assert!(
+            database_url.to_ascii_lowercase().contains("test"),
+            "refusing to run destructive integration test outside a test database"
+        );
+        let redis_url = std::env::var("IMAGEFORGE_TEST_REDIS_URL")
+            .expect("IMAGEFORGE_TEST_REDIS_URL must be set");
+        let pool = PgPoolOptions::new()
+            .max_connections(16)
+            .connect(&database_url)
+            .await
+            .expect("connect quota test database");
+        sqlx::migrate!().run(&pool).await.expect("run migrations");
+        let state = build_test_state(pool.clone(), database_url, redis_url).await;
+        let marker = Uuid::new_v4().simple().to_string();
+        let mut cleanup = Vec::new();
+
+        let null_session = format!("quota-null-{marker}");
+        let null_ip: IpAddr = "198.51.100.11".parse().expect("parse fixture IP");
+        let null_date = reserve_anonymous_units(&state, &null_session, null_ip, 3)
+            .await
+            .expect("reserve NULL-rate batch quota");
+        let null_task = insert_anonymous_settlement_fixture(
+            &pool,
+            AnonymousSettlementFixture {
+                session_id: &null_session,
+                ip: null_ip,
+                date: null_date,
+                reserved_units: 3,
+                compression_rate: None,
+                total_files: 3,
+                completed_files: 3,
+            },
+        )
+        .await;
+        cleanup.push((null_task, null_session.clone(), null_ip, null_date));
+        assert_eq!(
+            settle_anonymous_task_reservation(&state, null_task)
+                .await
+                .expect("settle NULL-rate batch"),
+            Some(0)
+        );
+        assert_eq!(
+            anonymous_quota_counts(&state, &null_session, null_ip, null_date).await,
+            (3, 3)
+        );
+
+        let passthrough_session = format!("quota-passthrough-{marker}");
+        let passthrough_ip: IpAddr = "198.51.100.12".parse().expect("parse fixture IP");
+        let passthrough_date =
+            reserve_anonymous_units(&state, &passthrough_session, passthrough_ip, 3)
+                .await
+                .expect("reserve passthrough batch quota");
+        let passthrough_task = insert_anonymous_settlement_fixture(
+            &pool,
+            AnonymousSettlementFixture {
+                session_id: &passthrough_session,
+                ip: passthrough_ip,
+                date: passthrough_date,
+                reserved_units: 3,
+                compression_rate: Some(100),
+                total_files: 3,
+                completed_files: 3,
+            },
+        )
+        .await;
+        cleanup.push((
+            passthrough_task,
+            passthrough_session.clone(),
+            passthrough_ip,
+            passthrough_date,
+        ));
+        assert_eq!(
+            settle_anonymous_task_reservation(&state, passthrough_task)
+                .await
+                .expect("settle passthrough batch"),
+            Some(3)
+        );
+        assert_eq!(
+            anonymous_quota_counts(
+                &state,
+                &passthrough_session,
+                passthrough_ip,
+                passthrough_date,
+            )
+            .await,
+            (0, 0)
+        );
+
+        let partial_session = format!("quota-partial-{marker}");
+        let partial_ip: IpAddr = "198.51.100.13".parse().expect("parse fixture IP");
+        let partial_date = reserve_anonymous_units(&state, &partial_session, partial_ip, 3)
+            .await
+            .expect("reserve partial batch quota");
+        let partial_task = insert_anonymous_settlement_fixture(
+            &pool,
+            AnonymousSettlementFixture {
+                session_id: &partial_session,
+                ip: partial_ip,
+                date: partial_date,
+                reserved_units: 3,
+                compression_rate: None,
+                total_files: 3,
+                completed_files: 2,
+            },
+        )
+        .await;
+        cleanup.push((
+            partial_task,
+            partial_session.clone(),
+            partial_ip,
+            partial_date,
+        ));
+        assert_eq!(
+            settle_anonymous_task_reservation(&state, partial_task)
+                .await
+                .expect("settle partial batch"),
+            Some(1)
+        );
+        assert_eq!(
+            anonymous_quota_counts(&state, &partial_session, partial_ip, partial_date).await,
+            (2, 2)
+        );
+
+        let limit_session = format!("quota-limit-{marker}");
+        let limit_ip: IpAddr = "198.51.100.14".parse().expect("parse fixture IP");
+        let mut limit_date = None;
+        for batch in 0..2 {
+            let date = reserve_anonymous_units(&state, &limit_session, limit_ip, 5)
+                .await
+                .expect("reserve consecutive anonymous batch");
+            limit_date = Some(date);
+            let task_id = insert_anonymous_settlement_fixture(
+                &pool,
+                AnonymousSettlementFixture {
+                    session_id: &limit_session,
+                    ip: limit_ip,
+                    date,
+                    reserved_units: 5,
+                    compression_rate: None,
+                    total_files: 5,
+                    completed_files: 5,
+                },
+            )
+            .await;
+            cleanup.push((task_id, limit_session.clone(), limit_ip, date));
+            assert_eq!(
+                settle_anonymous_task_reservation(&state, task_id)
+                    .await
+                    .expect("settle consecutive anonymous batch"),
+                Some(0),
+                "batch {batch} unexpectedly refunded consumed units"
+            );
+        }
+        let limit_error = reserve_anonymous_units(&state, &limit_session, limit_ip, 1)
+            .await
+            .expect_err("daily anonymous quota was bypassed");
+        assert_eq!(limit_error.code, ErrorCode::QuotaExceeded);
+        assert_eq!(
+            anonymous_quota_counts(
+                &state,
+                &limit_session,
+                limit_ip,
+                limit_date.expect("limit quota date"),
+            )
+            .await,
+            (10, 10)
+        );
+
+        let mut redis = state.redis.clone();
+        for (task_id, session_id, ip, date) in cleanup {
+            sqlx::query("DELETE FROM tasks WHERE id = $1")
+                .bind(task_id)
+                .execute(&pool)
+                .await
+                .expect("delete quota settlement fixture");
+            let _: i64 = redis::cmd("DEL")
+                .arg(anonymous_session_key(&session_id, date))
+                .arg(anonymous_ip_key(ip, date))
+                .arg(format!("anon_quota_refund:{task_id}"))
+                .query_async(&mut redis)
+                .await
+                .expect("delete quota settlement Redis keys");
+        }
+    }
 }

src/worker/mod.rs

@@ -1086,11 +1086,14 @@ async fn process_task_file(
     } else {
         (original_size.saturating_sub(compressed_size) as f64) * 100.0 / (original_size as f64)
     };
-    let skip_charge = compression_rate == Some(100)
-        && format_in == format_out
-        && max_width.is_none()
-        && max_height.is_none();
-    let charge_units = !skip_charge && compressed_size < original_size;
+    let charge_units = quota::output_consumes_unit(
+        compression_rate,
+        format_in == format_out,
+        max_width.is_some() || max_height.is_some(),
+        false,
+        original_size,
+        compressed_size,
+    );
 
     let object_key = storage::result_attempt_key(
         ctx.retention_hours as i64,