Skip to main content

max / makenotwork

13.3 KB · 381 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 //! 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. The failed
100 /// subscription IDs are listed in the ticket body.
101 pub fn spawn_fan_sub_fanout(
102 bg: &BackgroundTx,
103 stripe: Arc<dyn PaymentProvider>,
104 account_id: StripeAccountId,
105 sub_ids: Vec<String>,
106 op: FanSubOp,
107 wam: Option<crate::wam_client::WamClient>,
108 ) {
109 if sub_ids.is_empty() {
110 return;
111 }
112 bg.spawn("fan-sub stripe fan-out", async move {
113 let total = sub_ids.len();
114 let mut failed_ids: Vec<&str> = Vec::new();
115 for sub_id in &sub_ids {
116 if let Err(e) = op.apply(&stripe, sub_id, account_id.as_str()).await {
117 failed_ids.push(sub_id);
118 tracing::warn!(stripe_sub_id = %sub_id, op = op.label(), error = ?e, "fan subscription op failed");
119 }
120 }
121 if failed_ids.is_empty() {
122 tracing::debug!(total, op = op.label(), "fan subscription fan-out completed");
123 return;
124 }
125 let failed = failed_ids.len();
126 tracing::warn!(total, failed, op = op.label(), "fan subscription fan-out completed with failures");
127 if let Some(wam) = wam {
128 let title = format!(
129 "Fan subscription fan-out incomplete: {} of {total} '{}' ops failed",
130 failed,
131 op.label()
132 );
133 let body = format!(
134 "Applying '{}' to a creator's fan subscriptions (account {}) left {failed} of {total} \
135 unreconciled. These subscriptions may still be charging (or retaining access) against \
136 the creator's current state and need manual reconciliation in Stripe.\n\nFailed subscription IDs:\n{}",
137 op.label(),
138 account_id.as_str(),
139 failed_ids.join("\n"),
140 );
141 wam.create_ticket(&title, Some(&body), "high", "fan-fanout-incomplete", Some(account_id.as_str())).await;
142 }
143 });
144 }
145
146 #[cfg(test)]
147 mod tests {
148 //! Operation labels. These reach logs and support tickets for creator
149 //! subscription changes, so a label that says the opposite of what happened
150 //! sends an investigation the wrong way.
151
152 use super::*;
153
154 #[test]
155 fn setting_and_clearing_cancel_at_period_end_do_not_share_a_label() {
156 assert_eq!(
157 FanSubOp::CancelAtPeriodEnd(true).label(),
158 "set_cancel_at_period_end"
159 );
160 assert_eq!(
161 FanSubOp::CancelAtPeriodEnd(false).label(),
162 "clear_cancel_at_period_end"
163 );
164 }
165
166 #[test]
167 fn every_operation_has_a_distinct_label() {
168 let labels = [
169 FanSubOp::CancelAtPeriodEnd(true).label(),
170 FanSubOp::CancelAtPeriodEnd(false).label(),
171 FanSubOp::Pause.label(),
172 FanSubOp::Resume.label(),
173 FanSubOp::Cancel.label(),
174 ];
175 let unique: std::collections::HashSet<_> = labels.iter().collect();
176 assert_eq!(unique.len(), labels.len(), "labels collide: {labels:?}");
177 }
178
179 // ── run_fan_sub_fanout ──
180 //
181 // The returned count is the whole contract: `mnw-admin` prints it, and the
182 // background twin turns the same count into a WAM ticket. A fan-out that
183 // reports zero failures when ops failed leaves fans charging against a
184 // creator state that no longer matches, with nothing to reconcile from.
185
186 use crate::payments::test_provider::ScriptedProvider;
187
188 /// Two handles on one provider: the fan-out takes the trait object, and the
189 /// assertions read the recorded calls off the concrete type.
190 fn provider(p: ScriptedProvider) -> (Arc<ScriptedProvider>, Arc<dyn PaymentProvider>) {
191 let scripted = Arc::new(p);
192 let stripe: Arc<dyn PaymentProvider> = scripted.clone();
193 (scripted, stripe)
194 }
195
196 fn account() -> StripeAccountId {
197 StripeAccountId::new("acct_test1234567890").expect("a well-formed connected account id")
198 }
199
200 fn subs(ids: &[&str]) -> Vec<String> {
201 ids.iter().map(|s| (*s).to_owned()).collect()
202 }
203
204 #[tokio::test]
205 async fn applies_the_op_to_every_subscription_in_order() {
206 let (scripted, stripe) = provider(ScriptedProvider::healthy());
207 let failed = run_fan_sub_fanout(
208 &stripe,
209 &account(),
210 &subs(&["sub_a", "sub_b", "sub_c"]),
211 FanSubOp::Pause,
212 )
213 .await;
214
215 assert_eq!(failed, 0);
216 assert_eq!(
217 scripted.calls(),
218 vec![
219 ("pause", "sub_a".to_string()),
220 ("pause", "sub_b".to_string()),
221 ("pause", "sub_c".to_string()),
222 ],
223 "every subscription gets the op, and none is skipped"
224 );
225 }
226
227 #[tokio::test]
228 async fn counts_the_failures_rather_than_reporting_success() {
229 let (_scripted, stripe) = provider(ScriptedProvider::failing(["sub_b", "sub_d"]));
230 let failed = run_fan_sub_fanout(
231 &stripe,
232 &account(),
233 &subs(&["sub_a", "sub_b", "sub_c", "sub_d"]),
234 FanSubOp::Cancel,
235 )
236 .await;
237
238 assert_eq!(
239 failed, 2,
240 "two ops failed; a count of 0 or 1 is a dropped cancel nobody reconciles"
241 );
242 }
243
244 #[tokio::test]
245 async fn a_failure_does_not_abort_the_rest() {
246 let (scripted, stripe) = provider(ScriptedProvider::failing(["sub_a"]));
247 let failed = run_fan_sub_fanout(
248 &stripe,
249 &account(),
250 &subs(&["sub_a", "sub_b", "sub_c"]),
251 FanSubOp::Resume,
252 )
253 .await;
254
255 assert_eq!(failed, 1);
256 assert_eq!(
257 scripted.calls().len(),
258 3,
259 "the first subscription failing must not strand the two behind it"
260 );
261 }
262
263 #[tokio::test]
264 async fn every_subscription_failing_is_counted_as_every_subscription() {
265 let (_scripted, stripe) = provider(ScriptedProvider::failing(["sub_a", "sub_b"]));
266 let failed = run_fan_sub_fanout(
267 &stripe,
268 &account(),
269 &subs(&["sub_a", "sub_b"]),
270 FanSubOp::Pause,
271 )
272 .await;
273
274 assert_eq!(failed, 2, "a saturating or decrementing counter shows here");
275 }
276
277 #[tokio::test]
278 async fn an_empty_list_is_no_calls_and_no_failures() {
279 let (scripted, stripe) = provider(ScriptedProvider::healthy());
280 let failed = run_fan_sub_fanout(&stripe, &account(), &[], FanSubOp::Cancel).await;
281
282 assert_eq!(failed, 0);
283 assert!(scripted.calls().is_empty());
284 }
285
286 #[tokio::test]
287 async fn cancel_at_period_end_carries_the_flag_it_was_given() {
288 for (cancel, expected) in [
289 (true, "set_cancel_at_period_end"),
290 (false, "clear_cancel_at_period_end"),
291 ] {
292 let (scripted, stripe) = provider(ScriptedProvider::healthy());
293 run_fan_sub_fanout(
294 &stripe,
295 &account(),
296 &subs(&["sub_a"]),
297 FanSubOp::CancelAtPeriodEnd(cancel),
298 )
299 .await;
300
301 assert_eq!(
302 scripted.calls(),
303 vec![(expected, "sub_a".to_string())],
304 "CancelAtPeriodEnd({cancel}) must reach Stripe as {expected}"
305 );
306 }
307 }
308
309 // ── spawn_fan_sub_fanout ──
310 //
311 // The background twin returns before doing anything, so the only thing a
312 // test can hold it to is that the work arrives. That is also the whole risk:
313 // a fan-out that silently never runs looks identical to one that succeeded,
314 // on a path whose failure is fans still being charged.
315
316 /// Wait for `want` calls to land, or give up. The queue is drained by a
317 /// separate task, so there is nothing to await on directly.
318 async fn settle(scripted: &Arc<ScriptedProvider>, want: usize) -> Vec<(&'static str, String)> {
319 for _ in 0..200 {
320 let calls = scripted.calls();
321 if calls.len() >= want {
322 return calls;
323 }
324 tokio::time::sleep(std::time::Duration::from_millis(5)).await;
325 }
326 scripted.calls()
327 }
328
329 #[tokio::test]
330 async fn the_queued_fan_out_actually_reaches_stripe() {
331 let (scripted, stripe) = provider(ScriptedProvider::healthy());
332 let bg = crate::background::spawn_pool_detached();
333
334 spawn_fan_sub_fanout(
335 &bg,
336 stripe,
337 account(),
338 subs(&["sub_a", "sub_b"]),
339 FanSubOp::Cancel,
340 None,
341 );
342
343 assert_eq!(
344 settle(&scripted, 2).await,
345 vec![
346 ("cancel", "sub_a".to_string()),
347 ("cancel", "sub_b".to_string()),
348 ],
349 "the ops must arrive; a fan-out that returns and does nothing is invisible"
350 );
351 }
352
353 #[tokio::test]
354 async fn a_failing_op_does_not_strand_the_queued_remainder() {
355 let (scripted, stripe) = provider(ScriptedProvider::failing(["sub_a"]));
356 let bg = crate::background::spawn_pool_detached();
357
358 spawn_fan_sub_fanout(
359 &bg,
360 stripe,
361 account(),
362 subs(&["sub_a", "sub_b", "sub_c"]),
363 FanSubOp::Pause,
364 None,
365 );
366
367 assert_eq!(settle(&scripted, 3).await.len(), 3);
368 }
369
370 #[tokio::test]
371 async fn an_empty_queued_fan_out_asks_stripe_for_nothing() {
372 let (scripted, stripe) = provider(ScriptedProvider::healthy());
373 let bg = crate::background::spawn_pool_detached();
374
375 spawn_fan_sub_fanout(&bg, stripe, account(), Vec::new(), FanSubOp::Resume, None);
376
377 tokio::time::sleep(std::time::Duration::from_millis(50)).await;
378 assert!(scripted.calls().is_empty());
379 }
380 }
381