Skip to main content

max / makenotwork

13.4 KB · 382 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 // ── run_fan_sub_fanout ──
181 //
182 // The returned count is the whole contract: `mnw-admin` prints it, and the
183 // background twin turns the same count into a WAM ticket. A fan-out that
184 // reports zero failures when ops failed leaves fans charging against a
185 // creator state that no longer matches, with nothing to reconcile from.
186
187 use crate::payments::test_provider::ScriptedProvider;
188
189 /// Two handles on one provider: the fan-out takes the trait object, and the
190 /// assertions read the recorded calls off the concrete type.
191 fn provider(p: ScriptedProvider) -> (Arc<ScriptedProvider>, Arc<dyn PaymentProvider>) {
192 let scripted = Arc::new(p);
193 let stripe: Arc<dyn PaymentProvider> = scripted.clone();
194 (scripted, stripe)
195 }
196
197 fn account() -> StripeAccountId {
198 StripeAccountId::new("acct_test1234567890").expect("a well-formed connected account id")
199 }
200
201 fn subs(ids: &[&str]) -> Vec<String> {
202 ids.iter().map(|s| (*s).to_owned()).collect()
203 }
204
205 #[tokio::test]
206 async fn applies_the_op_to_every_subscription_in_order() {
207 let (scripted, stripe) = provider(ScriptedProvider::healthy());
208 let failed = run_fan_sub_fanout(
209 &stripe,
210 &account(),
211 &subs(&["sub_a", "sub_b", "sub_c"]),
212 FanSubOp::Pause,
213 )
214 .await;
215
216 assert_eq!(failed, 0);
217 assert_eq!(
218 scripted.calls(),
219 vec![
220 ("pause", "sub_a".to_string()),
221 ("pause", "sub_b".to_string()),
222 ("pause", "sub_c".to_string()),
223 ],
224 "every subscription gets the op, and none is skipped"
225 );
226 }
227
228 #[tokio::test]
229 async fn counts_the_failures_rather_than_reporting_success() {
230 let (_scripted, stripe) = provider(ScriptedProvider::failing(["sub_b", "sub_d"]));
231 let failed = run_fan_sub_fanout(
232 &stripe,
233 &account(),
234 &subs(&["sub_a", "sub_b", "sub_c", "sub_d"]),
235 FanSubOp::Cancel,
236 )
237 .await;
238
239 assert_eq!(
240 failed, 2,
241 "two ops failed; a count of 0 or 1 is a dropped cancel nobody reconciles"
242 );
243 }
244
245 #[tokio::test]
246 async fn a_failure_does_not_abort_the_rest() {
247 let (scripted, stripe) = provider(ScriptedProvider::failing(["sub_a"]));
248 let failed = run_fan_sub_fanout(
249 &stripe,
250 &account(),
251 &subs(&["sub_a", "sub_b", "sub_c"]),
252 FanSubOp::Resume,
253 )
254 .await;
255
256 assert_eq!(failed, 1);
257 assert_eq!(
258 scripted.calls().len(),
259 3,
260 "the first subscription failing must not strand the two behind it"
261 );
262 }
263
264 #[tokio::test]
265 async fn every_subscription_failing_is_counted_as_every_subscription() {
266 let (_scripted, stripe) = provider(ScriptedProvider::failing(["sub_a", "sub_b"]));
267 let failed = run_fan_sub_fanout(
268 &stripe,
269 &account(),
270 &subs(&["sub_a", "sub_b"]),
271 FanSubOp::Pause,
272 )
273 .await;
274
275 assert_eq!(failed, 2, "a saturating or decrementing counter shows here");
276 }
277
278 #[tokio::test]
279 async fn an_empty_list_is_no_calls_and_no_failures() {
280 let (scripted, stripe) = provider(ScriptedProvider::healthy());
281 let failed = run_fan_sub_fanout(&stripe, &account(), &[], FanSubOp::Cancel).await;
282
283 assert_eq!(failed, 0);
284 assert!(scripted.calls().is_empty());
285 }
286
287 #[tokio::test]
288 async fn cancel_at_period_end_carries_the_flag_it_was_given() {
289 for (cancel, expected) in [
290 (true, "set_cancel_at_period_end"),
291 (false, "clear_cancel_at_period_end"),
292 ] {
293 let (scripted, stripe) = provider(ScriptedProvider::healthy());
294 run_fan_sub_fanout(
295 &stripe,
296 &account(),
297 &subs(&["sub_a"]),
298 FanSubOp::CancelAtPeriodEnd(cancel),
299 )
300 .await;
301
302 assert_eq!(
303 scripted.calls(),
304 vec![(expected, "sub_a".to_string())],
305 "CancelAtPeriodEnd({cancel}) must reach Stripe as {expected}"
306 );
307 }
308 }
309
310 // ── spawn_fan_sub_fanout ──
311 //
312 // The background twin returns before doing anything, so the only thing a
313 // test can hold it to is that the work arrives. That is also the whole risk:
314 // a fan-out that silently never runs looks identical to one that succeeded,
315 // on a path whose failure is fans still being charged.
316
317 /// Wait for `want` calls to land, or give up. The queue is drained by a
318 /// separate task, so there is nothing to await on directly.
319 async fn settle(scripted: &Arc<ScriptedProvider>, want: usize) -> Vec<(&'static str, String)> {
320 for _ in 0..200 {
321 let calls = scripted.calls();
322 if calls.len() >= want {
323 return calls;
324 }
325 tokio::time::sleep(std::time::Duration::from_millis(5)).await;
326 }
327 scripted.calls()
328 }
329
330 #[tokio::test]
331 async fn the_queued_fan_out_actually_reaches_stripe() {
332 let (scripted, stripe) = provider(ScriptedProvider::healthy());
333 let bg = crate::background::spawn_pool_detached();
334
335 spawn_fan_sub_fanout(
336 &bg,
337 stripe,
338 account(),
339 subs(&["sub_a", "sub_b"]),
340 FanSubOp::Cancel,
341 None,
342 );
343
344 assert_eq!(
345 settle(&scripted, 2).await,
346 vec![
347 ("cancel", "sub_a".to_string()),
348 ("cancel", "sub_b".to_string()),
349 ],
350 "the ops must arrive; a fan-out that returns and does nothing is invisible"
351 );
352 }
353
354 #[tokio::test]
355 async fn a_failing_op_does_not_strand_the_queued_remainder() {
356 let (scripted, stripe) = provider(ScriptedProvider::failing(["sub_a"]));
357 let bg = crate::background::spawn_pool_detached();
358
359 spawn_fan_sub_fanout(
360 &bg,
361 stripe,
362 account(),
363 subs(&["sub_a", "sub_b", "sub_c"]),
364 FanSubOp::Pause,
365 None,
366 );
367
368 assert_eq!(settle(&scripted, 3).await.len(), 3);
369 }
370
371 #[tokio::test]
372 async fn an_empty_queued_fan_out_asks_stripe_for_nothing() {
373 let (scripted, stripe) = provider(ScriptedProvider::healthy());
374 let bg = crate::background::spawn_pool_detached();
375
376 spawn_fan_sub_fanout(&bg, stripe, account(), Vec::new(), FanSubOp::Resume, None);
377
378 tokio::time::sleep(std::time::Duration::from_millis(50)).await;
379 assert!(scripted.calls().is_empty());
380 }
381 }
382