syrup_rail_postgres/
host_charge_reconciliation.rs

1use chrono::{DateTime, Utc};
2use sqlx::PgPool;
3use syrup_rail::{
4    BillingScopeId, GatewayAccountId, HostChargeTargetId, HostChargeTargetTransition,
5    HostChargeTargetTransitionKind, PaymentAttemptId, PaymentAttemptKind, SubscriberId,
6};
7use uuid::Uuid;
8
9use crate::{
10    attempts::LocalAttemptPolicy,
11    enrollment_application::set_application_timeouts,
12    host_charge_application::HostChargeApplicationError,
13    host_charges::HostChargeTargetStore,
14    reconciliation::{RECONCILIATION_CLAIM_RETRY_AFTER_SECONDS, RECONCILIATION_PHASE_BATCH_SIZE},
15};
16
17const STALE_UNSUBMITTED_HOST_CHARGE_TEXT: &str =
18    "Host charge was abandoned before gateway submission.";
19
20/// Outcome of one bounded stale host-charge cleanup page.
21///
22/// A skipped candidate keeps its financial and target state because its host
23/// target rejected the release transition, the attempt changed concurrently,
24/// or its row was contended. Its reconciliation claim timestamp advances so
25/// unclaimed work can progress before it is retried. The host target callback
26/// owns incident reporting for rejected transitions.
27#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
28pub struct StaleHostChargeCleanupSummary {
29    failed: u64,
30    skipped: u64,
31}
32
33impl StaleHostChargeCleanupSummary {
34    /// Attempts whose target release and local failure committed atomically.
35    pub const fn failed(self) -> u64 {
36        self.failed
37    }
38
39    /// Candidates left financially unchanged after rejection or revalidation.
40    pub const fn skipped(self) -> u64 {
41        self.skipped
42    }
43}
44
45#[derive(Clone, Copy)]
46struct StaleHostChargeCandidate {
47    attempt_id: Uuid,
48    billing_scope_id: Uuid,
49    subscriber_id: Uuid,
50    target_id: Uuid,
51}
52
53/// Fails one bounded account-scoped batch of stale local host charges.
54///
55/// Candidate selection first makes a durable scheduling claim with
56/// `FOR UPDATE SKIP LOCKED`. Previously claimed rows sort behind untouched work,
57/// so a bounded page cannot be monopolized by target-local skips. Each claimed
58/// candidate then transitions its host-owned target to
59/// `ReleasedBeforeSubmission` before locking and revalidating the canonical
60/// attempt. Both financial changes commit in one transaction. A concurrent
61/// submission, terminal outcome,
62/// target-level `StaleTarget` or `Unchanged`, or contended attempt row rolls the
63/// target transition back and counts as skipped. No gateway I/O is performed.
64pub async fn fail_stale_unsubmitted_host_charges(
65    pool: &PgPool,
66    targets: &dyn HostChargeTargetStore,
67    gateway_account_id: GatewayAccountId,
68) -> Result<StaleHostChargeCleanupSummary, HostChargeApplicationError> {
69    let policy = LocalAttemptPolicy::for_kind(PaymentAttemptKind::HostCharge);
70    let candidates = claim_stale_host_charge_candidates(pool, gateway_account_id).await?;
71
72    let mut summary = StaleHostChargeCleanupSummary::default();
73    for candidate in candidates {
74        let mut transaction = pool.begin().await?;
75        set_application_timeouts(&mut transaction).await?;
76        let effective_at: DateTime<Utc> = sqlx::query_scalar("SELECT clock_timestamp()")
77            .fetch_one(&mut *transaction)
78            .await?;
79        let target_outcome = targets
80            .apply_transition(
81                &mut transaction,
82                HostChargeTargetTransition::new(
83                    BillingScopeId::new(candidate.billing_scope_id),
84                    SubscriberId::new(candidate.subscriber_id),
85                    PaymentAttemptId::new(candidate.attempt_id),
86                    HostChargeTargetId::new(candidate.target_id),
87                    HostChargeTargetTransitionKind::ReleasedBeforeSubmission,
88                    effective_at,
89                ),
90            )
91            .await?;
92        if !target_outcome.is_applied() {
93            tracing::warn!(
94                target: "syrup_rail::host_charge_reconciliation",
95                billing_scope_id = %candidate.billing_scope_id,
96                subscriber_id = %candidate.subscriber_id,
97                attempt_id = %candidate.attempt_id,
98                target_id = %candidate.target_id,
99                ?target_outcome,
100                "host target refused stale unsubmitted charge release; leaving attempt unresolved"
101            );
102            transaction.rollback().await?;
103            summary.skipped += 1;
104            continue;
105        }
106
107        let result = sqlx::query(
108            r#"
109            UPDATE billing_payment_attempts
110            SET status = 'failed',
111                gateway_response_text = $6,
112                gateway_condition = COALESCE(gateway_condition, 'failed'),
113                resolved_at = COALESCE(resolved_at, clock_timestamp()),
114                updated_at = clock_timestamp()
115            WHERE id = $1 AND billing_scope_id = $2 AND subscriber_id = $3
116                AND host_charge_target_id = $4 AND gateway_account_id = $5
117                AND attempt_kind = 'host_charge'
118                AND status = ANY($7::text[])
119                AND submitted_at IS NULL
120                AND created_at <= clock_timestamp()
121                    - ($8::bigint * interval '1 second')
122            "#,
123        )
124        .bind(candidate.attempt_id)
125        .bind(candidate.billing_scope_id)
126        .bind(candidate.subscriber_id)
127        .bind(candidate.target_id)
128        .bind(gateway_account_id.as_uuid())
129        .bind(STALE_UNSUBMITTED_HOST_CHARGE_TEXT)
130        .bind(LocalAttemptPolicy::expirable_status_values())
131        .bind(policy.stale_after_seconds())
132        .execute(&mut *transaction)
133        .await;
134        let result = match result {
135            Ok(result) => result,
136            Err(error) if is_lock_not_available(&error) => {
137                transaction.rollback().await?;
138                summary.skipped += 1;
139                continue;
140            }
141            Err(error) => return Err(error.into()),
142        };
143        if result.rows_affected() == 0 {
144            transaction.rollback().await?;
145            summary.skipped += 1;
146            continue;
147        }
148        transaction.commit().await?;
149        summary.failed += 1;
150    }
151    Ok(summary)
152}
153
154async fn claim_stale_host_charge_candidates(
155    pool: &PgPool,
156    gateway_account_id: GatewayAccountId,
157) -> Result<Vec<StaleHostChargeCandidate>, sqlx::Error> {
158    let policy = LocalAttemptPolicy::for_kind(PaymentAttemptKind::HostCharge);
159    let mut transaction = pool.begin().await?;
160    set_application_timeouts(&mut transaction).await?;
161    let candidates = sqlx::query_as::<_, (Uuid, Uuid, Uuid, Uuid)>(
162        r#"
163        WITH candidate_attempts AS MATERIALIZED (
164            SELECT attempts.id AS attempt_id,
165                attempts.billing_scope_id,
166                attempts.subscriber_id,
167                attempts.host_charge_target_id AS target_id,
168                attempts.created_at,
169                attempts.updated_at AS claimed_order_at
170            FROM billing_payment_attempts AS attempts
171            WHERE attempts.gateway_account_id = $1
172                AND attempts.attempt_kind = 'host_charge'
173                AND attempts.status = ANY($2::text[])
174                AND attempts.submitted_at IS NULL
175                AND attempts.created_at <= clock_timestamp()
176                    - ($3::bigint * interval '1 second')
177                AND attempts.updated_at <= clock_timestamp()
178                    - ($4::bigint * interval '1 second')
179            ORDER BY attempts.updated_at, attempts.created_at, attempts.id
180            LIMIT $5
181            FOR UPDATE OF attempts SKIP LOCKED
182        ), claimed_attempts AS (
183            UPDATE billing_payment_attempts AS attempts
184            SET updated_at = clock_timestamp()
185            FROM candidate_attempts
186            WHERE attempts.id = candidate_attempts.attempt_id
187            RETURNING attempts.id
188        )
189        SELECT candidate_attempts.attempt_id,
190            candidate_attempts.billing_scope_id,
191            candidate_attempts.subscriber_id,
192            candidate_attempts.target_id
193        FROM candidate_attempts
194        INNER JOIN claimed_attempts
195            ON claimed_attempts.id = candidate_attempts.attempt_id
196        ORDER BY candidate_attempts.claimed_order_at,
197            candidate_attempts.created_at,
198            candidate_attempts.attempt_id
199        "#,
200    )
201    .bind(gateway_account_id.as_uuid())
202    .bind(LocalAttemptPolicy::expirable_status_values())
203    .bind(policy.stale_after_seconds())
204    .bind(RECONCILIATION_CLAIM_RETRY_AFTER_SECONDS)
205    .bind(RECONCILIATION_PHASE_BATCH_SIZE)
206    .fetch_all(&mut *transaction)
207    .await?
208    .into_iter()
209    .map(
210        |(attempt_id, billing_scope_id, subscriber_id, target_id)| StaleHostChargeCandidate {
211            attempt_id,
212            billing_scope_id,
213            subscriber_id,
214            target_id,
215        },
216    )
217    .collect::<Vec<_>>();
218    transaction.commit().await?;
219    Ok(candidates)
220}
221
222fn is_lock_not_available(error: &sqlx::Error) -> bool {
223    matches!(
224        error,
225        sqlx::Error::Database(error) if error.code().as_deref() == Some("55P03")
226    )
227}
228
229#[cfg(test)]
230mod tests;
syrup_rail_postgres/host_charge_reconciliation.rs

syrup_rail_postgres/
host_charge_reconciliation.rs
