Compare commits

..

6 Commits

Author SHA1 Message Date
237899745
03b3a08185 fix: bound large WebP lossless probe cost
Some checks failed
CI / verify (push) Has been cancelled
2026-07-26 12:22:53 +08:00
237899745
66454b6325 fix: honor target image size across compression paths
Some checks failed
CI / verify (push) Has been cancelled
2026-07-26 12:14:05 +08:00
237899745
72f36c631e fix(worker): qualify file attempt claim
Some checks failed
CI / verify (push) Has been cancelled
2026-07-26 11:06:53 +08:00
237899745
cdec19977c ci: require all external state tests
Some checks failed
CI / verify (push) Has been cancelled
2026-07-26 10:54:22 +08:00
237899745
408e09cda8 fix(storage): make queue and object ownership durable 2026-07-26 10:54:13 +08:00
237899745
f2d490edce fix(billing): enforce one effective subscription per user 2026-07-26 10:54:01 +08:00
25 changed files with 3660 additions and 722 deletions

View File

@@ -33,7 +33,7 @@ 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: '10'
EXPECTED_EXTERNAL_TESTS: '13'
steps:
- name: Checkout
env:

View File

@@ -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` |
处理约束:

View File

@@ -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">

View File

@@ -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">

View File

@@ -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');

View File

@@ -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;

View File

@@ -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;

View File

@@ -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;

View File

@@ -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);

View File

@@ -1,7 +1,7 @@
#!/usr/bin/env bash
set -Eeuo pipefail
EXPECTED_EXTERNAL_TESTS="${EXPECTED_EXTERNAL_TESTS:-10}"
EXPECTED_EXTERNAL_TESTS="${EXPECTED_EXTERNAL_TESTS:-13}"
WORK_DIR="$(mktemp -d)"
POSTGRES_CONTAINER=""
REDIS_CONTAINER=""

View File

@@ -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 = state
.db
let mut tx = pool
.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 {
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(),
},
}))
Ok((subscription_id, period_start, period_end))
}
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");
}
}

View File

@@ -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 1 FROM subscriptions
WHERE user_id = $1 AND provider = 'stripe' AND status <> 'canceled'
)
SELECT provider
FROM subscriptions
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 {
return Err(AppError::new(
ErrorCode::IdempotencyConflict,
"已有 Stripe 订阅,请通过账单门户升级、降级或续费",
));
if let Some(provider) = open_subscription_provider {
let message = if provider == "stripe" {
"已有 Stripe 订阅,请通过账单门户升级、降级或续费"
} else {
"当前已有有效套餐,请在套餐结束后创建 Stripe 订阅"
};
return Err(AppError::new(ErrorCode::IdempotencyConflict, message));
}
sqlx::query(

View File

@@ -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,15 +273,35 @@ 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 mut anonymous_reservation_date = None;
let op: Result<CompressResponse, AppError> = (async {
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 {
QuotaContext::User(billing) => ensure_quota_available(&state, billing, 1).await?,
QuotaContext::ApiKey(billing, _) => ensure_quota_available(&state, billing, 1).await?,
@@ -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 =
storage::store_bytes(&state, &object_key, compressed, format_out.content_type())
.await?;
let tracked = object_lifecycle::store_tracked_bytes(
&state,
task_id,
Some(file_id),
"result",
&object_key,
compressed,
format_out.content_type(),
)
.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);
}
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 {
let response = 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),
},
})
})
.await;
};
let idem_completion = build_idempotency_completion(
idem_acquired,
operation_idem_owner,
idempotency_scope,
idempotency_key.as_deref(),
request_hash.as_deref(),
&response,
)?;
match op {
Ok(resp) => {
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(
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, "sync result commit response was lost; recovered committed publication");
}
Ok(false) => {
if let Err(cleanup_err) = object_lifecycle::schedule_tracked_delete(
&state,
scope,
idem_key,
request_hash,
200,
serde_json::to_value(&resp).unwrap_or(serde_json::Value::Null),
&tracked,
Some(&err),
)
.await;
.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);
}
}
Ok((
jar,
Json(Envelope {
success: true,
data: resp,
}),
))
}
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)
});
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,13 +616,33 @@ async fn compress_direct(
"请求正在处理中,请稍后重试",
));
}
idempotency::BeginResult::Acquired => {
idempotency::BeginResult::Acquired { owner } => {
idem_acquired = true;
idem_owner = Some(owner);
}
}
}
let op: Result<(axum::response::Response, DirectIdempotencyData), AppError> = (async {
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?,
QuotaContext::ApiKey(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 response = direct_response(compressed, format_out, &idem_data);
Ok((response, idem_data))
})
.await;
let idem_completion = build_idempotency_completion(
idem_acquired,
operation_idem_owner,
idempotency_scope,
idempotency_key.as_deref(),
request_hash.as_deref(),
&idem_data,
)?;
match op {
Ok((response, idem_data)) => {
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(
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,
scope,
idem_key,
request_hash,
200,
serde_json::to_value(&idem_data).unwrap_or(serde_json::Value::Null),
&tracked,
Some(&err),
)
.await;
.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);
}
}
Ok((jar, response))
}
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((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 最小为 10241KB",
));
}
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, $14,
NOW(), NOW(), $15, $16
$13, 1, 1, 0,
$14, $15,
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");
}
}

View File

@@ -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;
@@ -63,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)
@@ -289,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)
@@ -451,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 (
@@ -474,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)
@@ -515,6 +517,7 @@ async fn renew_zip_build_for(
AND zip_build_token = $2
AND zip_storage_key IS NULL
AND expires_at > NOW()
AND deletion_started_at IS NULL
"#,
)
.bind(task_id)
@@ -711,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)
})?;
@@ -766,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
@@ -789,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)
@@ -798,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 {
backend: stored.backend,
endpoint_id: stored.endpoint_id,
key: stored.key,
}),
Ok(result) if result.rows_affected() == 1 => {
if let Err(err) = object_lifecycle::publish_in_tx(&mut tx, &tracked).await {
tx.rollback().await.ok();
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))
}
}
@@ -825,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 {
backend: stored.backend.clone(),
endpoint_id: stored.endpoint_id,
key: stored.key.clone(),
};
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");
if let Err(err) = object_lifecycle::schedule_tracked_delete(state, tracked, None).await {
tracing::error!(task_id = %task_id, zip_build_token = %token, object_key = %tracked.stored.key, error = %err, "failed to persist unpublished ZIP deletion");
} else if let Err(err) = object_lifecycle::cleanup_ready_objects(state, 1, Some(task_id)).await
{
tracing::warn!(task_id = %task_id, zip_build_token = %token, object_key = %tracked.stored.key, error = %err, "unpublished ZIP deletion deferred");
}
release_zip_build(state, task_id, token).await;
}
@@ -856,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)
@@ -1247,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",
@@ -1268,13 +1283,20 @@ mod tests {
let orphan_read = storage::read_bytes(
&state,
&storage::ObjectLocator {
backend: unpublished.backend,
endpoint_id: unpublished.endpoint_id,
key: unpublished.key,
backend: unpublished.stored.backend,
endpoint_id: unpublished.stored.endpoint_id,
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");
let cancelled_task = Uuid::new_v4();
insert_zip_task(
@@ -1415,6 +1437,20 @@ mod tests {
.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)

View File

@@ -17,18 +17,33 @@ 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>(
"SELECT COUNT(*) FROM tasks WHERE status IN ('pending', 'processing')",
sqlx::query_as::<_, (i64, i64, i64, i64, i64)>(
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 {
Ok(Ok(value)) => (1, value),
_ => (0, 0),
};
let (database_up, active_tasks, outbox_pending, outbox_dead, delete_pending, staging) =
match database {
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))) => {
(1, 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!(

View File

@@ -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))
}

View File

@@ -8,8 +8,9 @@ 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::services::task_queue;
use crate::state::AppState;
use axum::extract::{ConnectInfo, Multipart, Path, State};
@@ -55,6 +56,7 @@ struct BatchFileInput {
struct BatchOptions {
level: CompressionLevel,
compression_rate: Option<u8>,
target_size_bytes: Option<u64>,
output_format: Option<ImageFmt>,
max_width: Option<u32>,
max_height: Option<u32>,
@@ -127,6 +129,7 @@ async fn create_batch_task(
}
let mut idem_acquired = false;
let mut idem_owner = None;
if let (Some(scope), Some(idem_key)) = (idempotency_scope, idempotency_key.as_deref()) {
let begin_result = idempotency::begin(
&state,
@@ -181,8 +184,9 @@ async fn create_batch_task(
"请求正在处理中,请稍后重试",
));
}
idempotency::BeginResult::Acquired => {
idempotency::BeginResult::Acquired { owner } => {
idem_acquired = true;
idem_owner = Some(owner);
}
}
}
@@ -190,8 +194,6 @@ async fn create_batch_task(
let mut anonymous_reserved_units = 0u32;
let mut anonymous_quota_date = None;
let mut task_persisted = false;
let mut enqueue_failure_finalized = false;
let mut cleanup_inputs_on_error = true;
let create_result: Result<BatchCreateResponse, AppError> = (async {
match &admission.task_owner {
TaskOwner::Anonymous { session_id } => {
@@ -234,16 +236,16 @@ async fn create_batch_task(
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,
expires_at, retention_hours, anonymous_units_reserved, anonymous_quota_date
) VALUES (
$1, $2, $3, $4, $5::inet, $6::task_source, 'pending',
$7::compression_level, $8, $9, $10, $11, $12,
$13, 0, 0,
$14, 0,
$15, $16, $17, $18
$13, $14, 0, 0,
$15, 0,
$16, $17, $18, $19
)
"#,
)
@@ -257,8 +259,9 @@ async fn create_batch_task(
.bind(opts.output_format.map(|f| f.as_str()))
.bind(opts.max_width.map(|v| v as i32))
.bind(opts.max_height.map(|v| v as i32))
.bind(false)
.bind(opts.preserve_metadata)
.bind(opts.compression_rate.map(|v| v as i16))
.bind(opts.target_size_bytes.map(|v| v as i64))
.bind(files.len() as i32)
.bind(total_original_size)
.bind(expires_at)
@@ -299,75 +302,67 @@ async fn create_batch_task(
})?;
}
sqlx::query("INSERT INTO task_queue_outbox (task_id) VALUES ($1)")
.bind(task_id)
.execute(&mut *tx)
.await
.map_err(|err| {
AppError::new(ErrorCode::Internal, "创建任务队列 outbox 失败").with_source(err)
})?;
let response = BatchCreateResponse {
task_id,
total_files: files.len() as i32,
status: "pending".to_string(),
status_url: format!("/api/v1/compress/tasks/{task_id}"),
};
if let (Some(scope), Some(idem_key)) = (idempotency_scope, idempotency_key.as_deref()) {
if idem_acquired {
idempotency::complete_in_tx(
&mut tx,
scope,
idem_key,
&request_hash,
idem_owner.ok_or_else(|| {
AppError::new(ErrorCode::Internal, "幂等操作租约缺失")
})?,
200,
serde_json::to_value(&response).map_err(|err| {
AppError::new(ErrorCode::Internal, "序列化幂等响应失败").with_source(err)
})?,
)
.await?;
}
}
tx.commit()
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "提交事务失败").with_source(err))?;
task_persisted = true;
if let Err(err) = enqueue_task(&state, task_id).await {
match finalize_enqueue_failure(&state, task_id, "队列提交失败").await {
Ok(true) => enqueue_failure_finalized = true,
Ok(false) => {
// XADD may have succeeded even if the client saw an error. A worker
// that already claimed the task owns both the input and settlement.
cleanup_inputs_on_error = false;
}
Err(finalize_err) => {
cleanup_inputs_on_error = false;
tracing::error!(task_id = %task_id, error = %finalize_err, "failed to finalize task after queue submission error");
}
let dispatch_state = state.clone();
tokio::spawn(async move {
if let Err(err) = task_queue::dispatch_task(&dispatch_state, task_id).await {
tracing::warn!(task_id = %task_id, error = %err, "immediate task queue dispatch deferred to outbox loop");
}
return Err(err);
}
});
Ok(BatchCreateResponse {
task_id,
total_files: files.len() as i32,
status: "pending".to_string(),
status_url: format!("/api/v1/compress/tasks/{task_id}"),
})
Ok(response)
})
.await;
match create_result {
Ok(resp) => {
if let (Some(scope), Some(idem_key)) = (idempotency_scope, idempotency_key.as_deref()) {
if idem_acquired {
let _ = idempotency::complete(
&state,
scope,
idem_key,
&request_hash,
200,
serde_json::to_value(&resp).unwrap_or(serde_json::Value::Null),
)
.await;
}
}
Ok((
jar,
Json(Envelope {
success: true,
data: resp,
}),
))
}
Ok(resp) => Ok((
jar,
Json(Envelope {
success: true,
data: resp,
}),
)),
Err(err) => {
if anonymous_reserved_units > 0 {
if let context::Principal::Anonymous { session_id } = &principal {
let should_refund_directly = if enqueue_failure_finalized {
match quota::settle_anonymous_task_reservation(&state, task_id).await {
Ok(Some(_)) => false,
Ok(None) => true,
Err(settle_err) => {
tracing::warn!(task_id = %task_id, error = %settle_err, "failed to settle anonymous batch admission");
false
}
}
} else {
!task_persisted
};
if should_refund_directly {
if !task_persisted {
if let Some(date) = anonymous_quota_date {
if let Err(refund_err) = quota::refund_anonymous_reservation_once(
&state,
@@ -387,10 +382,13 @@ async fn create_batch_task(
}
if let (Some(scope), Some(idem_key)) = (idempotency_scope, idempotency_key.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;
}
}
}
if cleanup_inputs_on_error {
if !task_persisted {
cleanup_task_input_dir(&state, task_id).await;
}
Err(err)
@@ -484,107 +482,6 @@ fn plan_upload_limits(plan: &Plan) -> Result<BatchUploadLimits, AppError> {
})
}
async fn enqueue_task(state: &AppState, task_id: Uuid) -> Result<(), AppError> {
let mut conn = state.redis.clone();
let now = Utc::now().to_rfc3339();
redis::cmd("XADD")
.arg("stream:compress_jobs")
.arg("MAXLEN")
.arg("~")
.arg(100_000)
.arg("*")
.arg("task_id")
.arg(task_id.to_string())
.arg("created_at")
.arg(now)
.query_async::<_, redis::Value>(&mut conn)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "写入队列失败").with_source(err))?;
Ok(())
}
async fn finalize_enqueue_failure(
state: &AppState,
task_id: Uuid,
error_message: &str,
) -> Result<bool, AppError> {
let mut tx = state.db.begin().await.map_err(|err| {
AppError::new(ErrorCode::Internal, "开启队列失败收口事务失败").with_source(err)
})?;
let task: Option<(String, i32)> =
sqlx::query_as("SELECT status::text, total_files FROM tasks WHERE id = $1 FOR UPDATE")
.bind(task_id)
.fetch_optional(&mut *tx)
.await
.map_err(|err| {
AppError::new(ErrorCode::Internal, "锁定队列失败任务失败").with_source(err)
})?;
let Some((status, total_files)) = task else {
tx.rollback().await.ok();
return Ok(false);
};
if status != "pending" {
tx.rollback().await.ok();
return Ok(false);
}
let files = sqlx::query(
r#"
UPDATE task_files
SET status = 'failed',
error_message = $2,
completed_at = NOW(),
input_path = NULL,
storage_path = NULL,
lease_owner = NULL,
lease_until = NULL
WHERE task_id = $1
AND status = 'pending'
"#,
)
.bind(task_id)
.bind(error_message)
.execute(&mut *tx)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "终结未入队文件失败").with_source(err))?;
if files.rows_affected() != u64::try_from(total_files.max(0)).unwrap_or(0) {
tx.rollback().await.ok();
return Err(AppError::new(
ErrorCode::Internal,
"未入队任务的文件状态不一致",
));
}
let task = sqlx::query(
r#"
UPDATE tasks
SET status = 'failed',
error_message = $2,
completed_at = NOW(),
completed_files = 0,
failed_files = total_files,
lease_owner = NULL,
lease_until = NULL
WHERE id = $1
AND status = 'pending'
"#,
)
.bind(task_id)
.bind(error_message)
.execute(&mut *tx)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "终结未入队任务失败").with_source(err))?;
if task.rows_affected() != 1 {
tx.rollback().await.ok();
return Ok(false);
}
tx.commit().await.map_err(|err| {
AppError::new(ErrorCode::Internal, "提交队列失败收口事务失败").with_source(err)
})?;
Ok(true)
}
async fn parse_batch_request(
state: &AppState,
task_id: Uuid,
@@ -596,6 +493,7 @@ async fn parse_batch_request(
let mut opts = BatchOptions {
level: CompressionLevel::Medium,
compression_rate: None,
target_size_bytes: None,
output_format: None,
max_width: None,
max_height: None,
@@ -786,6 +684,36 @@ async fn parse_batch_request(
});
}
}
"target_size_bytes" | "target_size" => {
let v = text.trim();
if !v.is_empty() {
let parsed = match v.parse::<u64>() {
Ok(parsed) if parsed >= 1024 && i64::try_from(parsed).is_ok() => parsed,
Ok(parsed) if parsed >= 1024 => {
cleanup_file_paths(&files).await;
return Err(AppError::new(
ErrorCode::InvalidRequest,
"target_size_bytes 超出支持范围",
));
}
Ok(_) => {
cleanup_file_paths(&files).await;
return Err(AppError::new(
ErrorCode::InvalidRequest,
"target_size_bytes 最小为 10241KB",
));
}
Err(_) => {
cleanup_file_paths(&files).await;
return Err(AppError::new(
ErrorCode::InvalidRequest,
"target_size_bytes 格式错误,需为正整数(字节)",
));
}
};
opts.target_size_bytes = Some(parsed);
}
}
"max_width" => {
let v = text.trim();
if !v.is_empty() {
@@ -826,6 +754,14 @@ async fn parse_batch_request(
}
}
if opts.compression_rate.is_some() && opts.target_size_bytes.is_some() {
cleanup_file_paths(&files).await;
return Err(AppError::new(
ErrorCode::InvalidRequest,
"compression_rate 与 target_size_bytes 不能同时指定",
));
}
if let Some(rate) = opts.compression_rate {
opts.level = compress::rate_to_level(rate);
}
@@ -836,6 +772,22 @@ async fn parse_batch_request(
}
}
if opts.target_size_bytes.is_some() {
if let Some(file) = files
.iter()
.find(|file| !compress::supports_target_size_format(file.output_format))
{
cleanup_file_paths(&files).await;
return Err(AppError::new(
ErrorCode::InvalidRequest,
format!(
"target_size_bytes 仅支持输出 jpeg/webp/avif当前为 {}",
file.output_format.as_str()
),
));
}
}
let mw = opts.max_width.map(|v| v.to_string()).unwrap_or_default();
let mh = opts.max_height.map(|v| v.to_string()).unwrap_or_default();
let out_fmt = opts.output_format.map(|f| f.as_str()).unwrap_or("");
@@ -843,6 +795,10 @@ async fn parse_batch_request(
.compression_rate
.map(|v| v.to_string())
.unwrap_or_default();
let target_key = opts
.target_size_bytes
.map(|v| v.to_string())
.unwrap_or_default();
let preserve = if opts.preserve_metadata { "1" } else { "0" };
let mut h = Sha256::new();
@@ -850,6 +806,7 @@ async fn parse_batch_request(
h.update(opts.level.as_str().as_bytes());
h.update(out_fmt.as_bytes());
h.update(rate_key.as_bytes());
h.update(target_key.as_bytes());
h.update(mw.as_bytes());
h.update(mh.as_bytes());
h.update(preserve.as_bytes());
@@ -948,7 +905,7 @@ async fn get_task(
user_id,
session_id
FROM tasks
WHERE id = $1
WHERE id = $1 AND deletion_started_at IS NULL
"#,
)
.bind(task_id)
@@ -1051,7 +1008,7 @@ async fn cancel_task(
context::require_api_permission(&principal, &["compress", "batch_compress"])?;
let task = sqlx::query_as::<_, TaskRow>(
"SELECT status::text AS status, total_files, completed_files, failed_files, created_at, completed_at, expires_at, user_id, session_id FROM tasks WHERE id = $1",
"SELECT status::text AS status, total_files, completed_files, failed_files, created_at, completed_at, expires_at, user_id, session_id FROM tasks WHERE id = $1 AND deletion_started_at IS NULL",
)
.bind(task_id)
.fetch_optional(&state.db)
@@ -1077,7 +1034,7 @@ async fn cancel_task(
}
let updated = sqlx::query(
"UPDATE tasks SET status = 'cancelled', completed_at = NOW() WHERE id = $1 AND status IN ('pending', 'processing')",
"UPDATE tasks SET status = 'cancelled', completed_at = NOW() WHERE id = $1 AND deletion_started_at IS NULL AND status IN ('pending', 'processing')",
)
.bind(task_id)
.execute(&state.db)
@@ -1116,13 +1073,16 @@ async fn delete_task(
let (jar, principal) = context::authenticate(&state, jar, &headers, ip).await?;
context::require_api_permission(&principal, &["compress", "batch_compress"])?;
let mut tx = state.db.begin().await.map_err(|err| {
AppError::new(ErrorCode::Internal, "开启任务删除事务失败").with_source(err)
})?;
let task = sqlx::query_as::<_, TaskRow>(
"SELECT status::text AS status, total_files, completed_files, failed_files, created_at, completed_at, expires_at, user_id, session_id FROM tasks WHERE id = $1",
"SELECT status::text AS status, total_files, completed_files, failed_files, created_at, completed_at, expires_at, user_id, session_id FROM tasks WHERE id = $1 FOR UPDATE",
)
.bind(task_id)
.fetch_optional(&state.db)
.fetch_optional(&mut *tx)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "查询任务失败").with_source(err))?
.map_err(|err| AppError::new(ErrorCode::Internal, "锁定待删除任务失败").with_source(err))?
.ok_or_else(|| AppError::new(ErrorCode::NotFound, "任务不存在"))?;
authorize_task(
@@ -1138,87 +1098,18 @@ async fn delete_task(
));
}
if task.status == "pending" {
let updated = sqlx::query(
"UPDATE tasks SET status = 'cancelled', completed_at = NOW() WHERE id = $1 AND status = 'pending'",
)
.bind(task_id)
.execute(&state.db)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "锁定待删除任务失败").with_source(err))?;
if updated.rows_affected() == 0 {
return Err(AppError::new(
ErrorCode::InvalidRequest,
"任务状态已变化,请刷新后重试",
));
}
if !object_lifecycle::mark_task_deleting(&mut tx, task_id, "user", false).await? {
return Err(AppError::new(
ErrorCode::InvalidRequest,
"任务状态已变化,请刷新后重试",
));
}
tx.commit().await.map_err(|err| {
AppError::new(ErrorCode::Internal, "提交任务删除状态失败").with_source(err)
})?;
quota::settle_anonymous_task_reservation(&state, task_id).await?;
let files = sqlx::query_as::<_, TaskStorageRow>(
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 = sqlx::query_as::<_, TaskZipStorageRow>(
"SELECT zip_storage_backend, zip_storage_endpoint_id, zip_storage_key FROM tasks WHERE id = $1",
)
.bind(task_id)
.fetch_one(&state.db)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "查询 ZIP 文件失败").with_source(err))?;
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;
let deleted = sqlx::query("DELETE FROM tasks WHERE id = $1")
.bind(task_id)
.execute(&state.db)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "删除任务失败").with_source(err))?;
if deleted.rows_affected() == 0 {
return Err(AppError::new(ErrorCode::NotFound, "任务不存在"));
if let Err(err) = object_lifecycle::finalize_task_deletion(&state, task_id).await {
tracing::warn!(task_id = %task_id, error = %err, "task deletion persisted and will be retried by lifecycle maintenance");
}
Ok((
@@ -1230,21 +1121,6 @@ async fn delete_task(
))
}
#[derive(Debug, FromRow)]
struct TaskStorageRow {
storage_backend: String,
storage_endpoint_id: Option<Uuid>,
storage_key: Option<String>,
input_path: Option<String>,
}
#[derive(Debug, FromRow)]
struct TaskZipStorageRow {
zip_storage_backend: Option<String>,
zip_storage_endpoint_id: Option<Uuid>,
zip_storage_key: Option<String>,
}
fn authorize_task(
principal: &context::Principal,
user_id: Option<Uuid>,
@@ -1334,7 +1210,7 @@ mod tests {
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[ignore = "requires isolated IMAGEFORGE_TEST_DATABASE_URL and IMAGEFORGE_TEST_REDIS_URL"]
async fn enqueue_failure_finalizes_files_and_removes_exact_input_directory() {
async fn outbox_retries_ambiguous_delivery_and_dead_letters_pending_task() {
let database_url = std::env::var("IMAGEFORGE_TEST_DATABASE_URL")
.expect("IMAGEFORGE_TEST_DATABASE_URL must be set");
assert!(
@@ -1410,6 +1286,11 @@ mod tests {
.await
.expect("insert pending batch file");
}
sqlx::query("INSERT INTO task_queue_outbox (task_id) VALUES ($1)")
.bind(task_id)
.execute(&pool)
.await
.expect("insert task outbox");
let mut redis = state.redis.clone();
let _: i64 = redis::cmd("DEL")
@@ -1424,17 +1305,48 @@ mod tests {
.await
.expect("install WRONGTYPE fixture");
let enqueue_error = enqueue_task(&state, task_id)
.await
.expect_err("XADD unexpectedly accepted a string key");
assert_eq!(enqueue_error.code, ErrorCode::Internal);
assert!(
finalize_enqueue_failure(&state, task_id, "队列提交失败")
task_queue::dispatch_task(&state, task_id)
.await
.expect("finalize enqueue failure"),
"pending task was not finalized"
.expect("dispatch outbox through WRONGTYPE"),
"outbox was not claimed"
);
let after_first_failure: (String, String, i32) = sqlx::query_as(
r#"
SELECT task.status::text, outbox.status, outbox.attempts
FROM tasks AS task
JOIN task_queue_outbox AS outbox ON outbox.task_id = task.id
WHERE task.id = $1
"#,
)
.bind(task_id)
.fetch_one(&pool)
.await
.expect("query retryable outbox state");
assert_eq!(
after_first_failure,
("pending".to_string(), "pending".to_string(), 1)
);
assert!(
tokio::fs::try_exists(&input_dir)
.await
.expect("check retained input directory"),
"a retryable Redis error deleted task inputs"
);
sqlx::query(
"UPDATE task_queue_outbox SET attempts = 19, next_attempt_at = NOW() WHERE task_id = $1",
)
.bind(task_id)
.execute(&pool)
.await
.expect("advance outbox to final attempt");
assert!(
task_queue::dispatch_task(&state, task_id)
.await
.expect("dispatch final outbox attempt"),
"final outbox attempt was not claimed"
);
cleanup_task_input_dir(&state, task_id).await;
let task: (String, bool, i32, i32) = sqlx::query_as(
r#"
@@ -1459,8 +1371,22 @@ mod tests {
.expect("query finalized task files");
assert_eq!(files.len(), 2);
assert!(files.iter().all(|row| {
row.0 == "failed" && row.1 && row.2 && row.3.as_deref() == Some("队列提交失败")
row.0 == "failed"
&& row.1
&& row.2
&& row
.3
.as_deref()
.is_some_and(|message| message.starts_with("队列持续不可用:"))
}));
let outbox_status: (String, i32, bool) = sqlx::query_as(
"SELECT status, attempts, last_error IS NOT NULL FROM task_queue_outbox WHERE task_id = $1",
)
.bind(task_id)
.fetch_one(&pool)
.await
.expect("query dead outbox");
assert_eq!(outbox_status, ("dead".to_string(), 20, true));
assert!(
!tokio::fs::try_exists(&input_dir)
.await
@@ -1490,11 +1416,225 @@ mod tests {
.query_async(&mut redis)
.await
.expect("remove WRONGTYPE fixture");
sqlx::query("DELETE FROM tasks WHERE id = $1")
.bind(task_id)
let ambiguous_task = Uuid::new_v4();
sqlx::query(
r#"
INSERT INTO tasks (
id, session_id, status, total_files, expires_at, retention_hours
) VALUES ($1, $2, 'pending', 0, NOW() + INTERVAL '1 day', 24)
"#,
)
.bind(ambiguous_task)
.bind(format!("ambiguous-{marker}"))
.execute(&pool)
.await
.expect("insert ambiguous delivery task");
sqlx::query("INSERT INTO task_queue_outbox (task_id) VALUES ($1)")
.bind(ambiguous_task)
.execute(&pool)
.await
.expect("delete batch test task");
.expect("insert ambiguous delivery outbox");
task_queue::enqueue_task(&state, ambiguous_task)
.await
.expect("simulate XADD success with lost reply");
assert!(task_queue::dispatch_task(&state, ambiguous_task)
.await
.expect("retry ambiguous delivery"));
let ambiguous_stream_len: i64 = redis::cmd("XLEN")
.arg("stream:compress_jobs")
.query_async(&mut redis)
.await
.expect("count duplicate ambiguous messages");
assert_eq!(ambiguous_stream_len, 2);
sqlx::query("UPDATE tasks SET status = 'completed', completed_at = NOW() WHERE id = $1")
.bind(ambiguous_task)
.execute(&pool)
.await
.expect("simulate first duplicate message completing the task");
assert_eq!(
crate::worker::process_task(&state, ambiguous_task, Uuid::new_v4())
.await
.expect("process ambiguous delivery task"),
TaskProcessOutcome::Done
);
assert_eq!(
crate::worker::process_task(&state, ambiguous_task, Uuid::new_v4())
.await
.expect("reprocess duplicate ambiguous delivery task"),
TaskProcessOutcome::Done
);
let ambiguous_status: String =
sqlx::query_scalar("SELECT status::text FROM tasks WHERE id = $1")
.bind(ambiguous_task)
.fetch_one(&pool)
.await
.expect("query ambiguous task status");
assert_eq!(ambiguous_status, "completed");
let concurrent_task = Uuid::new_v4();
sqlx::query(
"INSERT INTO tasks (id, session_id, status, total_files, expires_at, retention_hours) VALUES ($1, $2, 'pending', 0, NOW() + INTERVAL '1 day', 24)",
)
.bind(concurrent_task)
.bind(format!("concurrent-{marker}"))
.execute(&pool)
.await
.expect("insert concurrent dispatcher task");
sqlx::query("INSERT INTO task_queue_outbox (task_id) VALUES ($1)")
.bind(concurrent_task)
.execute(&pool)
.await
.expect("insert concurrent dispatcher outbox");
let barrier = Arc::new(tokio::sync::Barrier::new(20));
let mut dispatchers = Vec::new();
for _ in 0..20 {
let state = state.clone();
let barrier = barrier.clone();
dispatchers.push(tokio::spawn(async move {
barrier.wait().await;
task_queue::dispatch_task(&state, concurrent_task).await
}));
}
let mut claimed = 0;
for dispatcher in dispatchers {
if dispatcher
.await
.expect("dispatcher task panicked")
.expect("concurrent dispatcher failed")
{
claimed += 1;
}
}
assert_eq!(claimed, 1, "multiple dispatchers owned one outbox row");
let takeover_task = Uuid::new_v4();
sqlx::query(
"INSERT INTO tasks (id, session_id, status, total_files, expires_at, retention_hours) VALUES ($1, $2, 'pending', 0, NOW() + INTERVAL '1 day', 24)",
)
.bind(takeover_task)
.bind(format!("takeover-{marker}"))
.execute(&pool)
.await
.expect("insert dispatcher takeover task");
sqlx::query(
r#"
INSERT INTO task_queue_outbox (
task_id, status, attempts, lease_owner, lease_until
) VALUES ($1, 'delivering', 1, $2, NOW() - INTERVAL '1 second')
"#,
)
.bind(takeover_task)
.bind(Uuid::new_v4())
.execute(&pool)
.await
.expect("insert expired dispatcher lease");
assert!(task_queue::dispatch_task(&state, takeover_task)
.await
.expect("take over expired dispatcher lease"));
let takeover_state: (String, i32) =
sqlx::query_as("SELECT status, attempts FROM task_queue_outbox WHERE task_id = $1")
.bind(takeover_task)
.fetch_one(&pool)
.await
.expect("query dispatcher takeover state");
assert_eq!(takeover_state, ("delivered".to_string(), 2));
let idem_user: Uuid = sqlx::query_scalar(
r#"
INSERT INTO users (email, username, password_hash, email_verified_at)
VALUES ($1, $2, 'test', NOW())
RETURNING id
"#,
)
.bind(format!("outbox-idem-{marker}@example.test"))
.bind(format!("outbox-idem-{marker}"))
.fetch_one(&pool)
.await
.expect("insert outbox idempotency user");
let idem_key = format!("outbox-{marker}");
let idem_hash = "b".repeat(64);
let idem_owner = match idempotency::begin(
&state,
idempotency::Scope::User(idem_user),
&idem_key,
&idem_hash,
24,
)
.await
.expect("acquire outbox idempotency key")
{
idempotency::BeginResult::Acquired { owner } => owner,
other => panic!("unexpected outbox idempotency begin result: {other:?}"),
};
let idem_task = Uuid::new_v4();
let idem_response = serde_json::json!({"task_id": idem_task});
let mut idem_tx = pool
.begin()
.await
.expect("begin atomic task/outbox response tx");
sqlx::query(
"INSERT INTO tasks (id, user_id, status, total_files, expires_at, retention_hours) VALUES ($1, $2, 'pending', 0, NOW() + INTERVAL '1 day', 24)",
)
.bind(idem_task)
.bind(idem_user)
.execute(&mut *idem_tx)
.await
.expect("insert idempotent task");
sqlx::query("INSERT INTO task_queue_outbox (task_id) VALUES ($1)")
.bind(idem_task)
.execute(&mut *idem_tx)
.await
.expect("insert idempotent task outbox");
idempotency::complete_in_tx(
&mut idem_tx,
idempotency::Scope::User(idem_user),
&idem_key,
&idem_hash,
idem_owner,
200,
idem_response.clone(),
)
.await
.expect("persist atomic idempotent task response");
idem_tx.commit().await.expect("commit idempotent task");
match idempotency::begin(
&state,
idempotency::Scope::User(idem_user),
&idem_key,
&idem_hash,
24,
)
.await
.expect("replay idempotent task")
{
idempotency::BeginResult::Replay { response_body } => {
assert_eq!(response_body, idem_response)
}
other => panic!("same idempotency key did not replay task: {other:?}"),
}
sqlx::query("DELETE FROM tasks WHERE id = ANY($1)")
.bind(vec![
task_id,
ambiguous_task,
concurrent_task,
takeover_task,
idem_task,
])
.execute(&pool)
.await
.expect("delete batch outbox test tasks");
sqlx::query("DELETE FROM users WHERE id = $1")
.bind(idem_user)
.execute(&pool)
.await
.expect("delete outbox idempotency user");
let _: i64 = redis::cmd("DEL")
.arg("stream:compress_jobs")
.query_async(&mut redis)
.await
.expect("clean compression stream");
let _ = tokio::fs::remove_dir_all(&storage_root).await;
}
}

View File

@@ -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 <> 'canceled'
AND provider_subscription_id <> $2
AND status IN ('active', 'trialing', 'past_due')
AND NOT (
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 = [

View File

@@ -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 WEBP_TARGET_MAX_QUALITY: u8 = 92;
const AVIF_TARGET_MAX_QUALITY: u8 = 90;
const JPEG_TARGET_MAX_QUALITY: u8 = 100;
const WEBP_TARGET_MAX_QUALITY: u8 = 100;
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| {

View File

@@ -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(
state: &AppState,
pub async fn complete_in_tx(
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 {
tracing::warn!("idempotency record not updated (maybe already completed?)");
if updated.rows_affected() != 1 {
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");
}
}

View File

@@ -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;

File diff suppressed because it is too large Load Diff

View File

@@ -330,27 +330,7 @@ fn retention_prefix(hours: i64) -> String {
}
}
pub async fn store_bytes<B>(
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(
pub(crate) 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(
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(
pub(crate) 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('\\')

367
src/services/task_queue.rs Normal file
View File

@@ -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()
}

View File

@@ -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>,
@@ -729,6 +720,7 @@ pub(crate) 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 @@ pub(crate) 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 @@ pub(crate) 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 @@ pub(crate) 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 process_task_file(
state: AppState,
async fn claim_task_file_attempt(
state: &AppState,
task_id: Uuid,
task_attempt: i64,
file_id: Uuid,
worker_id: Uuid,
file: TaskFileProcRow,
level: compress::CompressionLevel,
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(
) -> Result<Option<i64>, AppError> {
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;
}
Err(err) => {
let _ = storage::delete_object(&state, &stored_locator(&stored)).await;
mark_file_failed_and_cleanup(&state, &fence, &err.message, &input_path).await?;
discard_tracked_result(&state, &tracked, None).await;
}
Err(err) => match worker_result_was_committed(&state, &fence, &tracked).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(
"UPDATE tasks SET status = 'cancelled', completed_at = COALESCE(completed_at, NOW()) WHERE id = $1 AND expires_at < NOW() AND status IN ('pending', 'processing')",
)
.bind(task_id)
.execute(&state.db)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "终止过期任务失败").with_source(err))?;
quota::settle_anonymous_task_reservation(state, task_id).await?;
let files: Vec<CleanupFileRow> = sqlx::query_as(
r#"
SELECT storage_backend, storage_endpoint_id,
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;
}
if object_lifecycle::mark_expired_task(state, task_id).await? {
object_lifecycle::finalize_task_deletion(state, task_id).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!(winning_object.backend, expected_backend);
assert_eq!(stale_object.stored.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))
.await
.expect("delete stale attempt object");
assert!(storage::read_bytes(&state, &stored_locator(&stale_object))
.await
.is_err());
discard_tracked_result(&state, &stale_object, None).await;
assert!(
storage::read_bytes(&state, &stored_locator(&stale_object.stored))
.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)