Skip to main content

max / makenotwork

6.8 KB · 180 lines History Blame Raw
1 //! Background fan-out of a Stripe subscription operation across all of a
2 //! creator's fan subscriptions.
3 //!
4 //! Pausing, suspending, or terminating a creator requires one Stripe call per
5 //! fan subscription. Run inline on a request or webhook handler, that serial
6 //! fan-out ties the hot path up for the duration of N Stripe round-trips,
7 //! minutes for a large creator, and on the webhook path long enough to trip
8 //! Stripe's delivery timeout and trigger a retry storm that re-runs the loop
9 //! (ultra-fuzz Run 6 S1/S2/R6-Perf-M1). This module moves the loop onto the
10 //! bounded background queue so the handler returns immediately. The operations
11 //! are fire-and-forget, errors were only logged inline, and still are.
12
13 use std::sync::Arc;
14
15 use crate::background::BackgroundTx;
16 use crate::db::StripeAccountId;
17 use crate::payments::PaymentProvider;
18
19 /// Which Stripe subscription operation to apply to each fan subscription.
20 #[derive(Clone, Copy, Debug)]
21 pub enum FanSubOp {
22 /// Set (`true`) or clear (`false`) cancel_at_period_end, creator pause/unpause.
23 CancelAtPeriodEnd(bool),
24 /// Pause collection, admin suspend.
25 Pause,
26 /// Resume collection, admin unsuspend.
27 Resume,
28 /// Cancel outright, admin terminate.
29 Cancel,
30 }
31
32 impl FanSubOp {
33 fn label(self) -> &'static str {
34 match self {
35 FanSubOp::CancelAtPeriodEnd(true) => "set_cancel_at_period_end",
36 FanSubOp::CancelAtPeriodEnd(false) => "clear_cancel_at_period_end",
37 FanSubOp::Pause => "pause",
38 FanSubOp::Resume => "resume",
39 FanSubOp::Cancel => "cancel",
40 }
41 }
42
43 async fn apply(
44 self,
45 stripe: &Arc<dyn PaymentProvider>,
46 sub_id: &str,
47 account_id: &str,
48 ) -> crate::error::Result<()> {
49 match self {
50 FanSubOp::CancelAtPeriodEnd(c) => {
51 stripe.set_cancel_at_period_end(sub_id, account_id, c).await
52 }
53 FanSubOp::Pause => stripe.pause_subscription(sub_id, account_id).await,
54 FanSubOp::Resume => stripe.resume_subscription(sub_id, account_id).await,
55 FanSubOp::Cancel => stripe.cancel_subscription(sub_id, account_id).await,
56 }
57 }
58 }
59
60 /// Apply `op` to every subscription in `sub_ids`, awaiting each call inline.
61 ///
62 /// For contexts with no background queue, the `mnw-admin` CLI, where blocking on
63 /// the serial Stripe fan-out is fine because there is no request or webhook hot
64 /// path to protect. Per-subscription failures are logged and do not abort the
65 /// rest; returns the number that failed so the caller can surface it. No-op (and
66 /// returns 0) for an empty list.
67 pub async fn run_fan_sub_fanout(
68 stripe: &Arc<dyn PaymentProvider>,
69 account_id: &StripeAccountId,
70 sub_ids: &[String],
71 op: FanSubOp,
72 ) -> usize {
73 let mut failed = 0usize;
74 for sub_id in sub_ids {
75 if let Err(e) = op.apply(stripe, sub_id, account_id.as_str()).await {
76 failed += 1;
77 tracing::warn!(stripe_sub_id = %sub_id, op = op.label(), error = ?e, "fan subscription op failed (inline)");
78 }
79 }
80 if failed > 0 {
81 tracing::warn!(
82 total = sub_ids.len(),
83 failed,
84 op = op.label(),
85 "inline fan subscription fan-out completed with failures"
86 );
87 }
88 failed
89 }
90
91 /// Apply `op` to every subscription in `sub_ids` on the background queue.
92 /// Returns immediately; the loop runs off the request/webhook hot path.
93 /// Per-subscription failures are logged and do not abort the rest. No-op for an
94 /// empty list.
95 ///
96 /// A failed op is not merely logged: any failures open a WAM ticket (when `wam`
97 /// is configured) so a dropped pause/suspend/cancel, a fan still charged after a
98 /// creator pause, or a suspended creator's fans retaining access, is actively
99 /// surfaced for manual reconciliation rather than lost in the logs (audit Run 13
100 /// Resilience: fan-out had no dead-letter). The failed subscription IDs are
101 /// listed in the ticket body.
102 pub fn spawn_fan_sub_fanout(
103 bg: &BackgroundTx,
104 stripe: Arc<dyn PaymentProvider>,
105 account_id: StripeAccountId,
106 sub_ids: Vec<String>,
107 op: FanSubOp,
108 wam: Option<crate::wam_client::WamClient>,
109 ) {
110 if sub_ids.is_empty() {
111 return;
112 }
113 bg.spawn("fan-sub stripe fan-out", async move {
114 let total = sub_ids.len();
115 let mut failed_ids: Vec<&str> = Vec::new();
116 for sub_id in &sub_ids {
117 if let Err(e) = op.apply(&stripe, sub_id, account_id.as_str()).await {
118 failed_ids.push(sub_id);
119 tracing::warn!(stripe_sub_id = %sub_id, op = op.label(), error = ?e, "fan subscription op failed");
120 }
121 }
122 if failed_ids.is_empty() {
123 tracing::debug!(total, op = op.label(), "fan subscription fan-out completed");
124 return;
125 }
126 let failed = failed_ids.len();
127 tracing::warn!(total, failed, op = op.label(), "fan subscription fan-out completed with failures");
128 if let Some(wam) = wam {
129 let title = format!(
130 "Fan subscription fan-out incomplete: {} of {total} '{}' ops failed",
131 failed,
132 op.label()
133 );
134 let body = format!(
135 "Applying '{}' to a creator's fan subscriptions (account {}) left {failed} of {total} \
136 unreconciled. These subscriptions may still be charging (or retaining access) against \
137 the creator's current state and need manual reconciliation in Stripe.\n\nFailed subscription IDs:\n{}",
138 op.label(),
139 account_id.as_str(),
140 failed_ids.join("\n"),
141 );
142 wam.create_ticket(&title, Some(&body), "high", "fan-fanout-incomplete", Some(account_id.as_str())).await;
143 }
144 });
145 }
146
147 #[cfg(test)]
148 mod tests {
149 //! Operation labels. These reach logs and support tickets for creator
150 //! subscription changes, so a label that says the opposite of what happened
151 //! sends an investigation the wrong way.
152
153 use super::*;
154
155 #[test]
156 fn setting_and_clearing_cancel_at_period_end_do_not_share_a_label() {
157 assert_eq!(
158 FanSubOp::CancelAtPeriodEnd(true).label(),
159 "set_cancel_at_period_end"
160 );
161 assert_eq!(
162 FanSubOp::CancelAtPeriodEnd(false).label(),
163 "clear_cancel_at_period_end"
164 );
165 }
166
167 #[test]
168 fn every_operation_has_a_distinct_label() {
169 let labels = [
170 FanSubOp::CancelAtPeriodEnd(true).label(),
171 FanSubOp::CancelAtPeriodEnd(false).label(),
172 FanSubOp::Pause.label(),
173 FanSubOp::Resume.label(),
174 FanSubOp::Cancel.label(),
175 ];
176 let unique: std::collections::HashSet<_> = labels.iter().collect();
177 assert_eq!(unique.len(), labels.len(), "labels collide: {labels:?}");
178 }
179 }
180