fix: make Stripe subscription events monotonic

src/api/webhooks.rs

@@ -12,20 +12,22 @@ use chrono::{TimeZone, Utc};
 use hmac::{Hmac, Mac};
 use serde::Deserialize;
 use sha2::Sha256;
+use sqlx::{Postgres, Transaction};
 
 pub fn router() -> Router<AppState> {
     Router::new().route("/webhooks/stripe", post(stripe_webhook))
 }
 
-#[derive(Debug, Deserialize)]
+#[derive(Debug, Clone, Deserialize)]
 struct StripeEvent {
     id: String,
+    created: i64,
     #[serde(rename = "type")]
     type_: String,
     data: StripeEventData,
 }
 
-#[derive(Debug, Deserialize)]
+#[derive(Debug, Clone, Deserialize)]
 struct StripeEventData {
     object: serde_json::Value,
 }
@@ -99,7 +101,7 @@ async fn stripe_webhook(
         ));
     }
 
-    if let Err(err) = process_stripe_event(&state, &event).await {
+    if let Err(err) = process_claimed_stripe_event(&state, &event).await {
         let _ = sqlx::query(
             "UPDATE webhook_events SET status = 'failed', error_message = $2, processed_at = NULL WHERE provider = 'stripe' AND provider_event_id = $1 AND status = 'processing'",
         )
@@ -111,19 +113,64 @@ async fn stripe_webhook(
         return Err(err);
     }
 
-    let _ = sqlx::query(
-        "UPDATE webhook_events SET status = 'processed', processed_at = NOW() WHERE provider = 'stripe' AND provider_event_id = $1",
-    )
-    .bind(&event.id)
-    .execute(&state.db)
-    .await;
-
     Ok(Json(Envelope {
         success: true,
         data: serde_json::json!({ "status": "ok" }),
     }))
 }
 
+async fn process_claimed_stripe_event(
+    state: &AppState,
+    event: &StripeEvent,
+) -> Result<(), AppError> {
+    let mut tx = state.db.begin().await.map_err(|err| {
+        AppError::new(ErrorCode::Internal, "开启 Webhook 事务失败").with_source(err)
+    })?;
+    let status: Option<String> = sqlx::query_scalar(
+        r#"
+        SELECT status
+        FROM webhook_events
+        WHERE provider = 'stripe' AND provider_event_id = $1
+        FOR UPDATE
+        "#,
+    )
+    .bind(&event.id)
+    .fetch_optional(&mut *tx)
+    .await
+    .map_err(|err| AppError::new(ErrorCode::Internal, "锁定 Webhook 失败").with_source(err))?;
+    if status.as_deref() != Some("processing") {
+        return Err(AppError::new(
+            ErrorCode::IdempotencyConflict,
+            "Webhook 事件未处于可处理状态",
+        ));
+    }
+
+    process_stripe_event(&mut tx, event).await?;
+    let updated = sqlx::query(
+        r#"
+        UPDATE webhook_events
+        SET status = 'processed', processed_at = NOW(), error_message = NULL
+        WHERE provider = 'stripe'
+          AND provider_event_id = $1
+          AND status = 'processing'
+        "#,
+    )
+    .bind(&event.id)
+    .execute(&mut *tx)
+    .await
+    .map_err(|err| AppError::new(ErrorCode::Internal, "更新 Webhook 状态失败").with_source(err))?;
+    if updated.rows_affected() != 1 {
+        return Err(AppError::new(
+            ErrorCode::IdempotencyConflict,
+            "Webhook 处理租约已失效",
+        ));
+    }
+    tx.commit().await.map_err(|err| {
+        AppError::new(ErrorCode::Internal, "提交 Webhook 事务失败").with_source(err)
+    })?;
+    Ok(())
+}
+
 fn verify_stripe_signature(payload: &[u8], sig_header: &str, secret: &str) -> Result<(), AppError> {
     let mut timestamp: Option<i64> = None;
     let mut signatures = Vec::<String>::new();
@@ -186,24 +233,25 @@ fn secure_eq(a: &str, b: &str) -> bool {
     out == 0
 }
 
-async fn process_stripe_event(state: &AppState, event: &StripeEvent) -> Result<(), AppError> {
+async fn process_stripe_event(
+    tx: &mut Transaction<'_, Postgres>,
+    event: &StripeEvent,
+) -> Result<(), AppError> {
     match event.type_.as_str() {
         "checkout.session.completed" => {
-            map_checkout_session_completed(state, &event.data.object).await
+            map_checkout_session_completed(tx, &event.data.object).await
         }
         "customer.subscription.created" | "customer.subscription.updated" => {
-            upsert_subscription(state, &event.data.object).await
-        }
-        "customer.subscription.deleted" => cancel_subscription(state, &event.data.object).await,
-        "invoice.paid" | "invoice.payment_failed" => {
-            upsert_invoice(state, &event.data.object).await
+            upsert_subscription(tx, event, &event.data.object).await
         }
+        "customer.subscription.deleted" => cancel_subscription(tx, event, &event.data.object).await,
+        "invoice.paid" | "invoice.payment_failed" => upsert_invoice(tx, &event.data.object).await,
         _ => Ok(()),
     }
 }
 
 async fn map_checkout_session_completed(
-    state: &AppState,
+    tx: &mut Transaction<'_, Postgres>,
     object: &serde_json::Value,
 ) -> Result<(), AppError> {
     let customer_id = object
@@ -242,7 +290,7 @@ async fn map_checkout_session_completed(
     )
     .bind(user_id)
     .bind(customer_id)
-    .execute(&state.db)
+    .execute(&mut **tx)
     .await
     .map_err(|err| {
         AppError::new(ErrorCode::Internal, "更新 Stripe Customer 映射失败").with_source(err)
@@ -253,7 +301,7 @@ async fn map_checkout_session_completed(
             "SELECT billing_customer_id FROM users WHERE id = $1",
         )
         .bind(user_id)
-        .fetch_optional(&state.db)
+        .fetch_optional(&mut **tx)
         .await
         .map_err(|err| AppError::new(ErrorCode::Internal, "查询用户失败").with_source(err))?
         .flatten();
@@ -275,11 +323,72 @@ async fn map_checkout_session_completed(
     Ok(())
 }
 
-async fn upsert_subscription(state: &AppState, object: &serde_json::Value) -> Result<(), AppError> {
+fn subscription_event_rank(event_type: &str) -> i16 {
+    match event_type {
+        "customer.subscription.deleted" => 2,
+        "customer.subscription.updated" => 1,
+        _ => 0,
+    }
+}
+
+async fn claim_subscription_event(
+    tx: &mut Transaction<'_, Postgres>,
+    event: &StripeEvent,
+    provider_subscription_id: &str,
+    is_deleted: bool,
+) -> Result<bool, AppError> {
+    let claimed: Option<String> = sqlx::query_scalar(
+        r#"
+        INSERT INTO provider_object_event_watermarks (
+          provider, object_type, provider_object_id,
+          last_event_created, last_event_rank, last_event_id, is_deleted
+        ) VALUES (
+          'stripe', 'subscription', $1,
+          $2, $3, $4, $5
+        )
+        ON CONFLICT (provider, object_type, provider_object_id) DO UPDATE
+        SET last_event_created = EXCLUDED.last_event_created,
+            last_event_rank = EXCLUDED.last_event_rank,
+            last_event_id = EXCLUDED.last_event_id,
+            is_deleted = EXCLUDED.is_deleted,
+            updated_at = NOW()
+        WHERE (
+          EXCLUDED.last_event_created,
+          EXCLUDED.last_event_rank,
+          EXCLUDED.last_event_id
+        ) > (
+          provider_object_event_watermarks.last_event_created,
+          provider_object_event_watermarks.last_event_rank,
+          provider_object_event_watermarks.last_event_id
+        )
+        RETURNING provider_object_id
+        "#,
+    )
+    .bind(provider_subscription_id)
+    .bind(event.created)
+    .bind(subscription_event_rank(&event.type_))
+    .bind(&event.id)
+    .bind(is_deleted)
+    .fetch_optional(&mut **tx)
+    .await
+    .map_err(|err| {
+        AppError::new(ErrorCode::Internal, "更新 Stripe 订阅事件水位失败").with_source(err)
+    })?;
+    Ok(claimed.is_some())
+}
+
+async fn upsert_subscription(
+    tx: &mut Transaction<'_, Postgres>,
+    event: &StripeEvent,
+    object: &serde_json::Value,
+) -> Result<(), AppError> {
     let provider_subscription_id = object
         .get("id")
         .and_then(|v| v.as_str())
         .ok_or_else(|| AppError::new(ErrorCode::InvalidRequest, "subscription.id 缺失"))?;
+    if !claim_subscription_event(tx, event, provider_subscription_id, false).await? {
+        return Ok(());
+    }
     let provider_customer_id = object
         .get("customer")
         .and_then(|v| v.as_str())
@@ -320,7 +429,7 @@ async fn upsert_subscription(state: &AppState, object: &serde_json::Value) -> Re
     let user_id: Option<uuid::Uuid> =
         sqlx::query_scalar("SELECT id FROM users WHERE billing_customer_id = $1 LIMIT 1")
             .bind(provider_customer_id)
-            .fetch_optional(&state.db)
+            .fetch_optional(&mut **tx)
             .await
             .map_err(|err| AppError::new(ErrorCode::Internal, "查询用户失败").with_source(err))?;
 
@@ -332,7 +441,7 @@ async fn upsert_subscription(state: &AppState, object: &serde_json::Value) -> Re
     let plan_id: Option<uuid::Uuid> =
         sqlx::query_scalar("SELECT id FROM plans WHERE stripe_price_id = $1 LIMIT 1")
             .bind(price_id)
-            .fetch_optional(&state.db)
+            .fetch_optional(&mut **tx)
             .await
             .map_err(|err| AppError::new(ErrorCode::Internal, "查询套餐失败").with_source(err))?;
 
@@ -341,20 +450,34 @@ async fn upsert_subscription(state: &AppState, object: &serde_json::Value) -> Re
         return Ok(());
     };
 
-    let updated: Option<uuid::Uuid> = sqlx::query_scalar(
+    sqlx::query(
         r#"
-        UPDATE subscriptions
-        SET user_id = $1,
-            plan_id = $2,
-            status = $3::subscription_status,
-            current_period_start = $4,
-            current_period_end = $5,
-            cancel_at_period_end = $6,
-            provider = 'stripe',
-            provider_customer_id = $7,
+        INSERT INTO subscriptions (
+          user_id, plan_id, status,
+          current_period_start, current_period_end,
+          cancel_at_period_end, canceled_at,
+          provider, provider_customer_id, provider_subscription_id
+        ) VALUES (
+          $1, $2, $3::subscription_status,
+          $4, $5,
+          $6,
+          CASE WHEN $3 = 'canceled' THEN NOW() ELSE NULL END,
+          'stripe', $7, $8
+        )
+        ON CONFLICT (provider, provider_subscription_id) DO UPDATE
+        SET user_id = EXCLUDED.user_id,
+            plan_id = EXCLUDED.plan_id,
+            status = EXCLUDED.status,
+            current_period_start = EXCLUDED.current_period_start,
+            current_period_end = EXCLUDED.current_period_end,
+            cancel_at_period_end = EXCLUDED.cancel_at_period_end,
+            canceled_at = CASE
+              WHEN EXCLUDED.status = 'canceled'::subscription_status
+                THEN COALESCE(subscriptions.canceled_at, NOW())
+              ELSE NULL
+            END,
+            provider_customer_id = EXCLUDED.provider_customer_id,
             updated_at = NOW()
-        WHERE provider = 'stripe' AND provider_subscription_id = $8
-        RETURNING id
         "#,
     )
     .bind(user_id)
@@ -365,66 +488,141 @@ async fn upsert_subscription(state: &AppState, object: &serde_json::Value) -> Re
     .bind(cancel_at_period_end)
     .bind(provider_customer_id)
     .bind(provider_subscription_id)
-    .fetch_optional(&state.db)
+    .execute(&mut **tx)
     .await
-    .map_err(|err| AppError::new(ErrorCode::Internal, "更新订阅失败").with_source(err))?;
-
-    if updated.is_none() {
-        let _ = sqlx::query(
-            r#"
-            INSERT INTO subscriptions (
-              user_id, plan_id, status,
-              current_period_start, current_period_end,
-              cancel_at_period_end,
-              provider, provider_customer_id, provider_subscription_id
-            ) VALUES (
-              $1, $2, $3::subscription_status,
-              $4, $5,
-              $6,
-              'stripe', $7, $8
-            )
-            "#,
-        )
-        .bind(user_id)
-        .bind(plan_id)
-        .bind(mapped_status)
-        .bind(current_period_start)
-        .bind(current_period_end)
-        .bind(cancel_at_period_end)
-        .bind(provider_customer_id)
-        .bind(provider_subscription_id)
-        .execute(&state.db)
-        .await
-        .map_err(|err| AppError::new(ErrorCode::Internal, "创建订阅失败").with_source(err))?;
-    }
+    .map_err(|err| AppError::new(ErrorCode::Internal, "写入订阅失败").with_source(err))?;
 
     Ok(())
 }
 
-async fn cancel_subscription(state: &AppState, object: &serde_json::Value) -> Result<(), AppError> {
+async fn cancel_subscription(
+    tx: &mut Transaction<'_, Postgres>,
+    event: &StripeEvent,
+    object: &serde_json::Value,
+) -> Result<(), AppError> {
     let provider_subscription_id = object
         .get("id")
         .and_then(|v| v.as_str())
         .ok_or_else(|| AppError::new(ErrorCode::InvalidRequest, "subscription.id 缺失"))?;
+    if !claim_subscription_event(tx, event, provider_subscription_id, true).await? {
+        return Ok(());
+    }
+    let canceled_at = object
+        .get("canceled_at")
+        .and_then(|v| v.as_i64())
+        .unwrap_or(event.created);
+    let canceled_at = Utc
+        .timestamp_opt(canceled_at, 0)
+        .single()
+        .unwrap_or_else(Utc::now);
 
-    let _ = sqlx::query(
+    let updated = sqlx::query(
         r#"
         UPDATE subscriptions
         SET status = 'canceled',
             cancel_at_period_end = false,
-            canceled_at = NOW(),
+            canceled_at = $2,
             updated_at = NOW()
         WHERE provider = 'stripe' AND provider_subscription_id = $1
         "#,
     )
     .bind(provider_subscription_id)
-    .execute(&state.db)
-    .await;
+    .bind(canceled_at)
+    .execute(&mut **tx)
+    .await
+    .map_err(|err| AppError::new(ErrorCode::Internal, "取消订阅失败").with_source(err))?;
+
+    if updated.rows_affected() == 0 {
+        let Some(provider_customer_id) = object
+            .get("customer")
+            .and_then(|value| value.as_str())
+            .filter(|value| !value.trim().is_empty())
+        else {
+            tracing::warn!(subscription = %provider_subscription_id, "deleted stripe subscription has no customer mapping; watermark retained");
+            return Ok(());
+        };
+        let Some(price_id) = object
+            .pointer("/items/data/0/price/id")
+            .and_then(|value| value.as_str())
+            .or_else(|| {
+                object
+                    .pointer("/items/data/0/plan/id")
+                    .and_then(|value| value.as_str())
+            })
+        else {
+            tracing::warn!(subscription = %provider_subscription_id, "deleted stripe subscription has no price mapping; watermark retained");
+            return Ok(());
+        };
+        let user_id: Option<uuid::Uuid> =
+            sqlx::query_scalar("SELECT id FROM users WHERE billing_customer_id = $1 LIMIT 1")
+                .bind(provider_customer_id)
+                .fetch_optional(&mut **tx)
+                .await
+                .map_err(|err| {
+                    AppError::new(ErrorCode::Internal, "查询用户失败").with_source(err)
+                })?;
+        let plan_id: Option<uuid::Uuid> =
+            sqlx::query_scalar("SELECT id FROM plans WHERE stripe_price_id = $1 LIMIT 1")
+                .bind(price_id)
+                .fetch_optional(&mut **tx)
+                .await
+                .map_err(|err| {
+                    AppError::new(ErrorCode::Internal, "查询套餐失败").with_source(err)
+                })?;
+        let (Some(user_id), Some(plan_id)) = (user_id, plan_id) else {
+            tracing::warn!(subscription = %provider_subscription_id, "deleted stripe subscription is not mapped locally; watermark retained");
+            return Ok(());
+        };
+        let period_start = object
+            .get("current_period_start")
+            .and_then(|value| value.as_i64())
+            .and_then(|value| Utc.timestamp_opt(value, 0).single())
+            .unwrap_or(canceled_at);
+        let period_end = object
+            .get("current_period_end")
+            .and_then(|value| value.as_i64())
+            .and_then(|value| Utc.timestamp_opt(value, 0).single())
+            .unwrap_or(canceled_at);
+
+        sqlx::query(
+            r#"
+            INSERT INTO subscriptions (
+              user_id, plan_id, status,
+              current_period_start, current_period_end,
+              cancel_at_period_end, canceled_at,
+              provider, provider_customer_id, provider_subscription_id
+            ) VALUES (
+              $1, $2, 'canceled', $3, $4, false, $5,
+              'stripe', $6, $7
+            )
+            ON CONFLICT (provider, provider_subscription_id) DO UPDATE
+            SET status = 'canceled',
+                cancel_at_period_end = false,
+                canceled_at = EXCLUDED.canceled_at,
+                updated_at = NOW()
+            "#,
+        )
+        .bind(user_id)
+        .bind(plan_id)
+        .bind(period_start)
+        .bind(period_end)
+        .bind(canceled_at)
+        .bind(provider_customer_id)
+        .bind(provider_subscription_id)
+        .execute(&mut **tx)
+        .await
+        .map_err(|err| {
+            AppError::new(ErrorCode::Internal, "创建已取消订阅 tombstone 失败").with_source(err)
+        })?;
+    }
 
     Ok(())
 }
 
-async fn upsert_invoice(state: &AppState, object: &serde_json::Value) -> Result<(), AppError> {
+async fn upsert_invoice(
+    tx: &mut Transaction<'_, Postgres>,
+    object: &serde_json::Value,
+) -> Result<(), AppError> {
     let provider_invoice_id = object
         .get("id")
         .and_then(|v| v.as_str())
@@ -437,7 +635,7 @@ async fn upsert_invoice(state: &AppState, object: &serde_json::Value) -> Result<
     let user_id: Option<uuid::Uuid> =
         sqlx::query_scalar("SELECT id FROM users WHERE billing_customer_id = $1 LIMIT 1")
             .bind(provider_customer_id)
-            .fetch_optional(&state.db)
+            .fetch_optional(&mut **tx)
             .await
             .map_err(|err| AppError::new(ErrorCode::Internal, "查询用户失败").with_source(err))?;
 
@@ -511,7 +709,7 @@ async fn upsert_invoice(state: &AppState, object: &serde_json::Value) -> Result<
     .bind(period_end)
     .bind(paid_at)
     .bind(provider_invoice_id)
-    .execute(&state.db)
+    .execute(&mut **tx)
     .await
     .map_err(|err| AppError::new(ErrorCode::Internal, "更新发票失败").with_source(err))?;
 
@@ -543,7 +741,7 @@ async fn upsert_invoice(state: &AppState, object: &serde_json::Value) -> Result<
         .bind(hosted_invoice_url.as_deref())
         .bind(pdf_url.as_deref())
         .bind(paid_at)
-        .execute(&state.db)
+        .execute(&mut **tx)
         .await
         .map_err(|err| AppError::new(ErrorCode::Internal, "创建发票失败").with_source(err))?;
     }
@@ -585,6 +783,10 @@ fn map_invoice_status(status: &str) -> &'static str {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use sqlx::postgres::PgPoolOptions;
+    use std::sync::Arc;
+    use tokio::sync::Barrier;
+    use uuid::Uuid;
 
     #[test]
     fn truncate_preserves_utf8_boundaries() {
@@ -592,4 +794,250 @@ mod tests {
         assert_eq!(truncate("abc中文".to_string(), 5), "abc");
         assert_eq!(truncate("short".to_string(), 20), "short");
     }
+
+    fn subscription_event(
+        event_id: &str,
+        event_type: &str,
+        created: i64,
+        subscription_id: &str,
+        customer_id: &str,
+        price_id: &str,
+    ) -> StripeEvent {
+        let status = if event_type == "customer.subscription.deleted" {
+            "canceled"
+        } else {
+            "active"
+        };
+        StripeEvent {
+            id: event_id.to_string(),
+            created,
+            type_: event_type.to_string(),
+            data: StripeEventData {
+                object: serde_json::json!({
+                    "id": subscription_id,
+                    "customer": customer_id,
+                    "status": status,
+                    "current_period_start": 1_700_000_000_i64,
+                    "current_period_end": 1_702_592_000_i64,
+                    "cancel_at_period_end": false,
+                    "canceled_at": if status == "canceled" { Some(created) } else { None },
+                    "items": { "data": [{ "price": { "id": price_id } }] }
+                }),
+            },
+        }
+    }
+
+    async fn apply_test_event(pool: &sqlx::PgPool, event: &StripeEvent) {
+        let mut tx = pool.begin().await.expect("begin event transaction");
+        process_stripe_event(&mut tx, event)
+            .await
+            .expect("apply stripe event");
+        tx.commit().await.expect("commit stripe event");
+    }
+
+    async fn assert_canceled_once(pool: &sqlx::PgPool, subscription_id: &str) {
+        let rows: Vec<(String,)> = sqlx::query_as(
+            "SELECT status::text FROM subscriptions WHERE provider = 'stripe' AND provider_subscription_id = $1",
+        )
+        .bind(subscription_id)
+        .fetch_all(pool)
+        .await
+        .expect("query subscription");
+        assert_eq!(rows, vec![("canceled".to_string(),)]);
+
+        let watermark: (bool,) = sqlx::query_as(
+            "SELECT is_deleted FROM provider_object_event_watermarks WHERE provider = 'stripe' AND object_type = 'subscription' AND provider_object_id = $1",
+        )
+        .bind(subscription_id)
+        .fetch_one(pool)
+        .await
+        .expect("query subscription watermark");
+        assert!(watermark.0);
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    #[ignore = "requires an isolated IMAGEFORGE_TEST_DATABASE_URL containing 'test'"]
+    async fn subscription_events_are_monotonic_for_all_orders_and_concurrency() {
+        let database_url = std::env::var("IMAGEFORGE_TEST_DATABASE_URL")
+            .expect("IMAGEFORGE_TEST_DATABASE_URL must be set");
+        assert!(
+            database_url.to_ascii_lowercase().contains("test"),
+            "refusing to run destructive integration test outside a test database"
+        );
+        let pool = PgPoolOptions::new()
+            .max_connections(16)
+            .connect(&database_url)
+            .await
+            .expect("connect test database");
+        sqlx::migrate!().run(&pool).await.expect("run migrations");
+
+        let marker = Uuid::new_v4().simple().to_string();
+        let customer_id = format!("cus_test_{marker}");
+        let price_id = format!("price_test_{marker}");
+        let user_id = Uuid::new_v4();
+        let plan_id = Uuid::new_v4();
+        sqlx::query(
+            r#"
+            INSERT INTO users (id, email, username, password_hash, billing_customer_id)
+            VALUES ($1, $2, $3, 'test-only', $4)
+            "#,
+        )
+        .bind(user_id)
+        .bind(format!("stripe-{marker}@example.test"))
+        .bind(format!("stripe_{marker}"))
+        .bind(&customer_id)
+        .execute(&pool)
+        .await
+        .expect("insert test user");
+        sqlx::query(
+            r#"
+            INSERT INTO plans (
+              id, code, name, stripe_price_id,
+              included_units_per_period, max_file_size_mb,
+              max_files_per_batch, concurrency_limit, retention_days
+            ) VALUES ($1, $2, 'Stripe ordering test', $3, 10, 10, 10, 1, 1)
+            "#,
+        )
+        .bind(plan_id)
+        .bind(format!("stripe_test_{marker}"))
+        .bind(&price_id)
+        .execute(&pool)
+        .await
+        .expect("insert test plan");
+
+        let permutations = [
+            [0, 1, 2],
+            [0, 2, 1],
+            [1, 0, 2],
+            [1, 2, 0],
+            [2, 0, 1],
+            [2, 1, 0],
+        ];
+        for (case, order) in permutations.into_iter().enumerate() {
+            let subscription_id = format!("sub_{marker}_perm_{case}");
+            let events = [
+                subscription_event(
+                    &format!("evt_{marker}_{case}_created"),
+                    "customer.subscription.created",
+                    1_700_000_100,
+                    &subscription_id,
+                    &customer_id,
+                    &price_id,
+                ),
+                subscription_event(
+                    &format!("evt_{marker}_{case}_updated"),
+                    "customer.subscription.updated",
+                    1_700_000_200,
+                    &subscription_id,
+                    &customer_id,
+                    &price_id,
+                ),
+                subscription_event(
+                    &format!("evt_{marker}_{case}_deleted"),
+                    "customer.subscription.deleted",
+                    1_700_000_300,
+                    &subscription_id,
+                    &customer_id,
+                    &price_id,
+                ),
+            ];
+            for index in order {
+                apply_test_event(&pool, &events[index]).await;
+            }
+            assert_canceled_once(&pool, &subscription_id).await;
+        }
+
+        let same_second_id = format!("sub_{marker}_same_second");
+        for event in [
+            subscription_event(
+                &format!("evt_{marker}_same_deleted"),
+                "customer.subscription.deleted",
+                1_700_000_400,
+                &same_second_id,
+                &customer_id,
+                &price_id,
+            ),
+            subscription_event(
+                &format!("evt_{marker}_same_updated"),
+                "customer.subscription.updated",
+                1_700_000_400,
+                &same_second_id,
+                &customer_id,
+                &price_id,
+            ),
+            subscription_event(
+                &format!("evt_{marker}_same_created"),
+                "customer.subscription.created",
+                1_700_000_400,
+                &same_second_id,
+                &customer_id,
+                &price_id,
+            ),
+        ] {
+            apply_test_event(&pool, &event).await;
+        }
+        assert_canceled_once(&pool, &same_second_id).await;
+
+        for case in 0..12 {
+            let subscription_id = format!("sub_{marker}_concurrent_{case}");
+            let events = [
+                subscription_event(
+                    &format!("evt_{marker}_concurrent_{case}_created"),
+                    "customer.subscription.created",
+                    1_700_001_100,
+                    &subscription_id,
+                    &customer_id,
+                    &price_id,
+                ),
+                subscription_event(
+                    &format!("evt_{marker}_concurrent_{case}_updated"),
+                    "customer.subscription.updated",
+                    1_700_001_200,
+                    &subscription_id,
+                    &customer_id,
+                    &price_id,
+                ),
+                subscription_event(
+                    &format!("evt_{marker}_concurrent_{case}_deleted"),
+                    "customer.subscription.deleted",
+                    1_700_001_300,
+                    &subscription_id,
+                    &customer_id,
+                    &price_id,
+                ),
+            ];
+            let barrier = Arc::new(Barrier::new(events.len()));
+            let mut joins = Vec::new();
+            for event in events {
+                let pool = pool.clone();
+                let barrier = barrier.clone();
+                joins.push(tokio::spawn(async move {
+                    barrier.wait().await;
+                    apply_test_event(&pool, &event).await;
+                }));
+            }
+            for join in joins {
+                join.await.expect("concurrent event task");
+            }
+            assert_canceled_once(&pool, &subscription_id).await;
+        }
+
+        sqlx::query(
+            "DELETE FROM provider_object_event_watermarks WHERE provider_object_id LIKE $1",
+        )
+        .bind(format!("sub_{marker}%"))
+        .execute(&pool)
+        .await
+        .expect("delete test watermarks");
+        sqlx::query("DELETE FROM users WHERE id = $1")
+            .bind(user_id)
+            .execute(&pool)
+            .await
+            .expect("delete test user");
+        sqlx::query("DELETE FROM plans WHERE id = $1")
+            .bind(plan_id)
+            .execute(&pool)
+            .await
+            .expect("delete test plan");
+    }
 }