fix(billing): stack gifts from paid period end after trial ends (#3406)

fluxer_api/src/api/stripe/StripeGiftTrialMetadata.ts

@@ -9,7 +9,7 @@ export function getGiftTrialMetadataKey(giftCode: string): string {
 	return `${GIFT_TRIAL_KEY_PREFIX}${giftCode}`;
 }
 
-function isGiftTrialRunning(subscription: Stripe.Subscription, nowMs: number): boolean {
+export function isGiftTrialRunning(subscription: Stripe.Subscription, nowMs: number): boolean {
 	return subscription.trial_end != null && subscription.trial_end * 1000 > nowMs;
 }
 

fluxer_api/src/api/stripe/services/StripeSubscriptionService.ts

@@ -10,7 +10,11 @@ import {addGiftCodeDuration} from '@app/api/models/GiftCode';
 import type {User} from '@app/api/models/User';
 import type {StoreEntitlementService} from '@app/api/store_billing/StoreEntitlementService';
 import type {ProductInfo, RecurringBillingCycle} from '@app/api/stripe/ProductRegistry';
-import {buildGiftTrialMetadata, getGiftTrialPaidUntil} from '@app/api/stripe/StripeGiftTrialMetadata';
+import {
+	buildGiftTrialMetadata,
+	getGiftTrialPaidUntil,
+	isGiftTrialRunning,
+} from '@app/api/stripe/StripeGiftTrialMetadata';
 import {
 	getPrimarySubscriptionItem,
 	getSubscriptionEntitlementPeriodEndUnix,
@@ -1293,7 +1297,8 @@ export class StripeSubscriptionService {
 			const currentTrialEnd = subscription.trial_end;
 			const item = getPrimarySubscriptionItem(subscription);
 			const currentPeriodEnd = getSubscriptionItemPeriodEndUnix(item);
-			const baseUnix = currentTrialEnd ?? currentPeriodEnd;
+			const nowMs = Date.now();
+			const baseUnix = isGiftTrialRunning(subscription, nowMs) ? currentTrialEnd : currentPeriodEnd;
 			Logger.debug(
 				{
 					userId: user.id,
@@ -1331,7 +1336,7 @@ export class StripeSubscriptionService {
 				{
 					trial_end: newTrialEndUnix,
 					proration_behavior: 'none',
-					metadata: buildGiftTrialMetadata(subscription, idempotencyKey, baseUnix, newTrialEndUnix, Date.now()),
+					metadata: buildGiftTrialMetadata(subscription, idempotencyKey, baseUnix, newTrialEndUnix, nowMs),
 				},
 				{idempotencyKey: stripeIdempotencyKey},
 			);

fluxer_api/src/api/stripe/tests/StripeGiftSubscriptionShift.test.ts

@@ -749,5 +749,45 @@ describe('gift time around a Stripe subscription', () => {
 			const afterWebhook = await findUser(account.userId);
 			expect(afterWebhook.premiumGiftExtensionEndsAt?.getTime()).toBe(trialEnd.getTime());
 		});
+
+		test('stacks from the paid period end when an earlier gift trial has ended', async () => {
+			const subscriptionId = 'sub_shift_stale_trial';
+			const periodEnd = Math.floor((Date.now() + ms('20 days')) / 1000);
+			const staleTrialEnd = periodEnd - 60 * 24 * 60 * 60;
+			useStripe({
+				subscriptions: {
+					[subscriptionId]: {
+						current_period_start: periodEnd - 30 * 24 * 60 * 60,
+						current_period_end: periodEnd,
+						trial_end: staleTrialEnd,
+						metadata: giftTrialMetadata('OLDGIFT', 30, staleTrialEnd - 30 * 24 * 60 * 60),
+					},
+				},
+			});
+			const {account} = await createUser({
+				premium_type: UserPremiumTypes.SUBSCRIPTION,
+				stripe_subscription_id: subscriptionId,
+				stripe_customer_id: 'cus_test_1',
+				premium_billing_cycle: 'monthly',
+				premium_until: new Date(periodEnd * 1000),
+			});
+			const gifter = await createTestAccount(harness);
+			await createBuilder(harness, gifter.token)
+				.post('/test/gifts/SHIFTSTALEGIFT')
+				.body({duration_type: 'days', duration_quantity: 30, created_by_user_id: gifter.userId})
+				.execute();
+
+			await createBuilder(harness, account.token).post('/gifts/SHIFTSTALEGIFT/redeem').expect(204).execute();
+
+			const update = stripeHandlers.spies.updatedSubscriptions[0]?.params;
+			expect(Number(update?.trial_end)).toBe(periodEnd + 30 * 24 * 60 * 60);
+			expect(update?.metadata).toEqual({
+				gtrial_OLDGIFT: '',
+				gtrial_paid_until: String(periodEnd),
+				gtrial_SHIFTSTALEGIFT: String(30 * 24 * 60 * 60),
+			});
+			const redeemed = await findUser(account.userId);
+			expect(redeemed.premiumUntil?.getTime()).toBe((periodEnd + 30 * 24 * 60 * 60) * 1000);
+		});
 	});
 });

fluxer_app/src/app/router/routes/RootRoutes.tsx

@@ -9,6 +9,13 @@ import {getDefaultLandingPath} from '@app/features/navigation/utils/DefaultLandi
 import {createRootRoute, createRoute} from '@app/features/platform/components/router/RouterBuilder';
 import {Redirect} from '@app/features/platform/components/router/RouterTypes';
 
+const enterDefaultLanding = afterSessionInitialization(() => {
+	if (readAuthenticatedRuntimeContext() === null) {
+		return new Redirect(Routes.LOGIN);
+	}
+	return new Redirect(getDefaultLandingPath());
+});
+
 export const rootRoute = createRootRoute({
 	layout: ({children}) => (
 		<RootComponent data-flx="app.router.root-routes.layout.root-component">{children}</RootComponent>
@@ -23,16 +30,11 @@ export const homeRoute = createRoute({
 	getParentRoute: () => rootRoute,
 	id: 'home',
 	path: '/',
-	onEnter: afterSessionInitialization(() => {
-		if (readAuthenticatedRuntimeContext() === null) {
-			return new Redirect(Routes.LOGIN);
-		}
-		return new Redirect(getDefaultLandingPath());
-	}),
+	onEnter: enterDefaultLanding,
 });
 export const appRoute = createRoute({
 	getParentRoute: () => rootRoute,
 	id: 'app',
 	path: Routes.APP,
-	onEnter: () => new Redirect(getDefaultLandingPath()),
+	onEnter: enterDefaultLanding,
 });

fluxer_app/src/features/voice/components/useVoiceCallTracksAndLayout.ts

@@ -468,9 +468,11 @@ export function useVoiceCallTracksAndLayout({channel, expandedUserIds}: UseVoice
 	useEffect(() => {
 		if (!pinnedParticipantIdentity) return;
 		if (cameraTracksAll.length === 0) return;
-		const stillExists = cameraTracksAll.some((tr) => tr.participant.identity === pinnedParticipantIdentity);
-		if (!stillExists) VoiceCallLayoutCommands.setPinnedParticipant(null);
-	}, [cameraTracksAll, pinnedParticipantIdentity]);
+		if (cameraTracksAll.some((tr) => tr.participant.identity === pinnedParticipantIdentity)) return;
+		const {connectionId} = parseVoiceParticipantIdentity(pinnedParticipantIdentity);
+		if (connectionVoiceStates[connectionId]?.channel_id === channel.id) return;
+		VoiceCallLayoutCommands.setPinnedParticipant(null);
+	}, [cameraTracksAll, channel.id, connectionVoiceStates, pinnedParticipantIdentity]);
 	const {
 		mainTrack: focusMainTrack,
 		carouselTracks,

fluxer_app/src/features/voice/engine/MediaEngineFacade.ts

@@ -801,6 +801,7 @@ class MediaEngineFacade extends Store {
 		restoreStream: boolean;
 		screenShareSnapshot: ScreenShareReconnectSnapshot | null;
 		toastWhenStreamNotRestored: boolean;
+		sendUpdateWhenStreamNotRestored: boolean;
 	}): Promise<void> {
 		assert.equal(typeof args.restoreCamera, 'boolean', 'restoreLocalMedia requires a camera flag');
 		assert.equal(typeof args.restoreStream, 'boolean', 'restoreLocalMedia requires a stream flag');
@@ -828,7 +829,9 @@ class MediaEngineFacade extends Store {
 		}
 		if (restored) return;
 		if (streamRestoreFailed || args.screenShareSnapshot !== null) {
-			voiceEngineV2AppMediaStateAdapter.applyScreenShareState(false, {sendUpdate: false});
+			voiceEngineV2AppMediaStateAdapter.applyScreenShareState(false, {
+				sendUpdate: args.sendUpdateWhenStreamNotRestored,
+			});
 		}
 		if (args.toastWhenStreamNotRestored && this.i18n) {
 			ToastCommands.createToast({
@@ -1690,7 +1693,7 @@ class MediaEngineFacade extends Store {
 				}
 				voiceEngineV2AppMediaExecutionAdapter.resetMicrophoneFailureLatch();
 				this.reconcileLocalAudioStateInBackground('region hot-swap complete');
-				if (isConversion) void this.reapplyVideoQualityLimits(newRoom);
+				if (isConversion) void this.finishMeshConversion(currentRoom, newRoom);
 			},
 			(_guildId, _channelId, _connectionId, _attemptId, error) => {
 				logger.warn('LiveKit connect failed, clearing gateway voice state', {error});
@@ -2715,6 +2718,19 @@ class MediaEngineFacade extends Store {
 		}
 	}
 
+	private async finishMeshConversion(meshRoom: Room, room: Room): Promise<void> {
+		await this.restoreLocalMedia({
+			reason: 'mesh conversion',
+			room,
+			restoreCamera: false,
+			restoreStream: false,
+			screenShareSnapshot: voiceEngineV2AppScreenShareExecutionAdapter.prepareScreenShareReconnect(meshRoom),
+			toastWhenStreamNotRestored: true,
+			sendUpdateWhenStreamNotRestored: true,
+		});
+		await this.reapplyVideoQualityLimits(room);
+	}
+
 	private async restoreLocalMediaState(roomOverride?: Room | null): Promise<void> {
 		const room = roomOverride ?? voiceEngineV2AppConnectionHostAdapter.room;
 		const participant = room?.localParticipant ?? null;
@@ -2728,6 +2744,7 @@ class MediaEngineFacade extends Store {
 			restoreStream: LocalVoiceState.getSelfStream() && !participant.isScreenShareEnabled,
 			screenShareSnapshot: pendingScreenShareReconnect,
 			toastWhenStreamNotRestored: false,
+			sendUpdateWhenStreamNotRestored: false,
 		});
 	}
 
@@ -2744,6 +2761,7 @@ class MediaEngineFacade extends Store {
 			restoreStream: pendingRestore.restoreStream,
 			screenShareSnapshot: null,
 			toastWhenStreamNotRestored: true,
+			sendUpdateWhenStreamNotRestored: false,
 		});
 	}
 

fluxer_app/src/features/voice/engine/VoiceRoomEventBinder.ts

@@ -284,6 +284,20 @@ export function bindRoomEvents(
 		participantSpeakingDisposers.get(identity)?.dispose();
 		participantSpeakingDisposers.delete(identity);
 	};
+	const observeRemoteScreenSharePublication = (participant: Participant, pub: RemoteTrackPublication): void => {
+		reportScreenSharePublicationObserved(participant.identity, pub.trackSid ?? null);
+		const streamKey = getStreamKeyForParticipantIdentity(guildId, channelId, participant.identity);
+		if (streamKey) {
+			cancelDeferredStopWatchingStreamKey(streamKey);
+		}
+	};
+	const observeExistingRemoteScreenSharePublications = (): void => {
+		room.remoteParticipants.forEach((participant) => {
+			for (const pub of participant.trackPublications.values()) {
+				if (pub.source === Track.Source.ScreenShare) observeRemoteScreenSharePublication(participant, pub);
+			}
+		});
+	};
 	const endScreenShareIfPublicationDidNotReturn = (): void => {
 		const publication = screenSharePublicationAwaitingReconnect;
 		screenSharePublicationAwaitingReconnect = null;
@@ -305,6 +319,7 @@ export function bindRoomEvents(
 			bindParticipantSpeakingEvents(room.localParticipant);
 			room.remoteParticipants.forEach((participant) => bindParticipantSpeakingEvents(participant));
 			dependencies.remoteSpeaking.hydrateFromRoom(room);
+			observeExistingRemoteScreenSharePublications();
 			bindScreenShareNegotiation();
 			dependencies.permissions.applyDeafen(room, getEffectiveAudioState().effectiveDeaf);
 			dependencies.connection.markConnected();
@@ -596,11 +611,7 @@ export function bindRoomEvents(
 					pub.setSubscribed(true);
 				}
 				if (pub.source === Track.Source.ScreenShare && !participant.isLocal) {
-					reportScreenSharePublicationObserved(participant.identity, pub.trackSid ?? null);
-					const streamKey = getStreamKeyForParticipantIdentity(guildId, channelId, participant.identity);
-					if (streamKey) {
-						cancelDeferredStopWatchingStreamKey(streamKey);
-					}
+					observeRemoteScreenSharePublication(participant, pub);
 					dependencies.subscriptions.reattachScreenShareAfterPublish(participant.identity, pub);
 					dependencies.subscriptions.reconcileSubscriptions();
 				}
@@ -615,6 +626,7 @@ export function bindRoomEvents(
 		}),
 	);
 	if (room.state === ConnectionState.Connected) {
+		guard(attemptId, observeExistingRemoteScreenSharePublications)();
 		guard(attemptId, bindScreenShareNegotiation)();
 	}
 }

fluxer_app/src/features/voice/engine/mesh/VoiceMeshEngine.ts

@@ -320,6 +320,7 @@ export class VoiceMeshEngine extends RTCEngine {
 	handlePeerStatus(peer: VoiceMeshPeer): void {
 		if (this.peers.get(peer.connectionId) !== peer) return;
 		const {status} = peer;
+		if (status === 'failed' && this.ownP2p() === false) return;
 		VoiceMeshPeers.setStatus(peer.connectionId, status);
 		if (status === 'connecting') return;
 		this.reportPeer(peer, status);
@@ -395,28 +396,30 @@ export class VoiceMeshEngine extends RTCEngine {
 		const serverMuted =
 			src === 'microphone' &&
 			voiceEngineV2AppVoiceStateAdapter.getVoiceStateByConnectionId(peer.connectionId)?.mute === true;
-		if (receiver && receiver.track.readyState === 'live' && publication.isDesired && !serverMuted) {
-			if (publication.track?.mediaStreamTrack !== receiver.track) {
-				participant.addSubscribedMediaTrack(receiver.track, info.sid, new MediaStream([receiver.track]), receiver);
-			}
-			return;
+		const live = receiver !== null && receiver.track.readyState === 'live' && publication.isDesired && !serverMuted;
+		if (live && publication.track?.mediaStreamTrack === receiver.track) return;
+		if (publication.track) {
+			publication.track.stop();
+			publication.setTrack(undefined);
+		}
+		if (live) {
+			participant.addSubscribedMediaTrack(receiver.track, info.sid, new MediaStream([receiver.track]), receiver);
 		}
-		if (!publication.track) return;
-		publication.track.stop();
-		publication.setTrack(undefined);
 	}
 
 	private readonly syncRoster = (): void => {
 		if (!this.room) return;
 		const {connectionId, channelId, guildId} = this.target;
-		const own = voiceEngineV2AppVoiceStateAdapter.getVoiceStateByConnectionId(connectionId);
-		const ownP2p = own?.channel_id === channelId ? own.p2p : undefined;
+		const ownP2p = this.ownP2p();
 		this.syncConversionWatchdog(ownP2p === false && !this.conversionHeld);
 		const roster = new Map<string, NormalizedVoiceState>();
-		if (ownP2p === true) {
+		if (ownP2p !== undefined) {
 			const states = voiceEngineV2AppVoiceStateAdapter.getAllVoiceStatesInChannel(guildId ?? ME, channelId);
 			for (const state of Object.values(states)) {
-				if (state.p2p === true && state.connection_id !== connectionId) roster.set(state.connection_id, state);
+				if (state.connection_id === connectionId) continue;
+				if (this.peers.has(state.connection_id) || (ownP2p && state.p2p === true)) {
+					roster.set(state.connection_id, state);
+				}
 			}
 		}
 		const before = this.peers.size;
@@ -431,6 +434,12 @@ export class VoiceMeshEngine extends RTCEngine {
 		for (const peer of this.peers.values()) this.syncPeerReceiver(peer, 'microphone');
 	};
 
+	private ownP2p(): boolean | undefined {
+		const {connectionId, channelId} = this.target;
+		const own = voiceEngineV2AppVoiceStateAdapter.getVoiceStateByConnectionId(connectionId);
+		return own?.channel_id === channelId ? own.p2p : undefined;
+	}
+
 	private addPeer(state: NormalizedVoiceState): void {
 		const peer = new VoiceMeshPeer(this, state.connection_id, state.user_id, this.target.connectionId);
 		this.peers.set(peer.connectionId, peer);

fluxer_app/src/features/voice/engine/v2/VoiceEngineV2AppConnectionHostAdapter.ts

@@ -3,7 +3,11 @@
 import assert from 'node:assert/strict';
 import Authentication from '@app/features/auth/state/Authentication';
 import {Logger} from '@app/features/platform/utils/AppLogger';
-import {createVoiceMeshRoom, holdVoiceMeshConversion} from '@app/features/voice/engine/mesh/VoiceMeshRoom';
+import {
+	createVoiceMeshRoom,
+	holdVoiceMeshConversion,
+	isVoiceMeshRoom,
+} from '@app/features/voice/engine/mesh/VoiceMeshRoom';
 import {Store} from '@app/features/voice/engine/Store';
 import {sendVoiceStateDisconnect} from '@app/features/voice/engine/VoiceChannelConnector';
 import {
@@ -785,8 +789,13 @@ export class VoiceEngineV2AppConnectionHostAdapter extends Store {
 			logger.warn('Region hot-swap: missing participant for track republishing');
 			return;
 		}
+		const isMeshConversion = isVoiceMeshRoom(oldRoom);
 		const publications = selectLocalMediaPublicationsForConnectionRepublish(
 			Array.from(oldParticipant.trackPublications.values()),
+		).filter(
+			(publication) =>
+				!isMeshConversion ||
+				(publication.source !== Track.Source.ScreenShare && publication.source !== Track.Source.ScreenShareAudio),
 		);
 		const errors: Array<{source: string; error: unknown}> = [];
 		let codecPolicyFailure: Error | null = null;

fluxer_media_proxy/src/constants.rs

@@ -16,7 +16,7 @@ pub const MAX_ANIMATED_FRAMES_DEFAULT: u32 = 20_000;
 pub const MAX_ANIMATED_TOTAL_PIXELS_DEFAULT: usize = 4 * MAX_MEDIA_IMAGE_PIXELS_DEFAULT;
 pub const MAX_JXL_DECODE_PIXELS: usize = 4096 * 4096;
 pub const MAX_STATIC_OUTPUT_PIXELS: usize = 6144 * 4096;
-pub const MAX_GIF_RESIZE_OUTPUT_PIXELS: usize = 16 << 20;
+pub const MAX_ANIMATED_OUTPUT_PIXELS: usize = 16 << 20;
 pub const SVG_RASTER_MAX_DIMENSION: u32 = 4096;
 pub const LARGE_SOURCE_LOG_PIXELS: usize = 4096 * 4096;
 const _: () = assert!(MAX_ANIMATED_FRAMES_DEFAULT >= 20_000);

fluxer_media_proxy/src/http_client/mod.rs

@@ -14,10 +14,15 @@ use std::time::Duration;
 
 pub type HttpClient = ClientWithMiddleware;
 
+const EXTERNAL_CONNECT_TIMEOUT_MS: u64 = 3_000;
+const EXTERNAL_READ_TIMEOUT_MS: u64 = 10_000;
+const EXTERNAL_RETRY_WINDOW_MS: u64 = 15_000;
+
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub struct HTTPClientOptions {
     connect_timeout_ms: NonZeroU64,
     request_timeout_ms: NonZeroU64,
+    read_timeout_ms: Option<NonZeroU64>,
     retries: HTTPRetryPolicy,
     address_policy: HTTPAddressPolicy,
 }
@@ -27,6 +32,7 @@ struct HTTPRetryPolicy {
     max_retries: NonZeroU32,
     min_delay_ms: NonZeroU64,
     max_delay_ms: NonZeroU64,
+    window_ms: Option<NonZeroU64>,
 }
 
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
@@ -44,6 +50,18 @@ impl HTTPClientOptions {
         }
     }
 
+    pub fn external(request_timeout_ms: NonZeroU64) -> Self {
+        let mut options = Self::new(
+            NonZeroU64::new(EXTERNAL_CONNECT_TIMEOUT_MS)
+                .expect("external connection timeout must be nonzero"),
+            request_timeout_ms,
+        )
+        .restrict_to_public();
+        options.read_timeout_ms = NonZeroU64::new(EXTERNAL_READ_TIMEOUT_MS);
+        options.retries.window_ms = NonZeroU64::new(EXTERNAL_RETRY_WINDOW_MS);
+        options
+    }
+
     pub fn restrict_to_public(mut self) -> Self {
         self.address_policy = HTTPAddressPolicy::PublicOnly;
         self
@@ -57,6 +75,7 @@ impl Default for HTTPClientOptions {
                 .expect("default connection timeout must be nonzero"),
             request_timeout_ms: NonZeroU64::new(30_000)
                 .expect("default request timeout must be nonzero"),
+            read_timeout_ms: None,
             retries: HTTPRetryPolicy {
                 max_retries: NonZeroU32::new(2)
                     .expect("default maximum retry count must be nonzero"),
@@ -64,6 +83,7 @@ impl Default for HTTPClientOptions {
                     .expect("default minimum retry delay must be nonzero"),
                 max_delay_ms: NonZeroU64::new(500)
                     .expect("default maximum retry delay must be nonzero"),
+                window_ms: None,
             },
             address_policy: HTTPAddressPolicy::Any,
         }
@@ -80,6 +100,9 @@ pub fn build_raw(options: HTTPClientOptions) -> Result<reqwest::Client, reqwest:
         .timeout(Duration::from_millis(options.request_timeout_ms.get()))
         .redirect(reqwest::redirect::Policy::none())
         .user_agent(crate::constants::OUTBOUND_USER_AGENT);
+    if let Some(read_timeout_ms) = options.read_timeout_ms {
+        builder = builder.read_timeout(Duration::from_millis(read_timeout_ms.get()));
+    }
     if options.address_policy == HTTPAddressPolicy::PublicOnly {
         builder = builder.dns_resolver(Arc::new(PinnedDnsResolver));
     }
@@ -96,6 +119,7 @@ pub fn build(
         max_retries,
         min_delay_ms,
         max_delay_ms,
+        window_ms,
     } = options.retries;
     let retry_policy = ExponentialBackoff::builder()
         .retry_bounds(
@@ -105,7 +129,11 @@ pub fn build(
         .build_with_max_retries(max_retries.get());
     Ok(builder
         .with(RetryTransientMiddleware::new_with_policy_and_strategy(
-            ObservableRetryPolicy::new(retry_policy, Arc::clone(&metrics)),
+            ObservableRetryPolicy::new(
+                retry_policy,
+                window_ms.map(|window| Duration::from_millis(window.get())),
+                Arc::clone(&metrics),
+            ),
             MediaProxyRetryStrategy::new(metrics),
         ))
         .build())
@@ -157,4 +185,24 @@ mod tests {
             .expect_err("the pinned resolver rejects a loopback host");
         assert!(crate::public_net_policy::is_pinned_dns_failure(&error));
     }
+
+    #[test]
+    fn the_external_client_fails_fast_on_slow_hosts() {
+        let options = HTTPClientOptions::external(millis(30_000));
+        assert_eq!(
+            EXTERNAL_CONNECT_TIMEOUT_MS,
+            options.connect_timeout_ms.get()
+        );
+        assert_eq!(30_000, options.request_timeout_ms.get());
+        assert_eq!(
+            Some(EXTERNAL_READ_TIMEOUT_MS),
+            options.read_timeout_ms.map(NonZeroU64::get)
+        );
+        assert_eq!(
+            Some(EXTERNAL_RETRY_WINDOW_MS),
+            options.retries.window_ms.map(NonZeroU64::get)
+        );
+        assert_eq!(HTTPAddressPolicy::PublicOnly, options.address_policy);
+        build_raw(options).expect("the external client configuration is valid");
+    }
 }

fluxer_media_proxy/src/http_client/retry.rs

@@ -8,22 +8,40 @@ use reqwest_retry::{
 };
 use std::io::ErrorKind;
 use std::sync::Arc;
-use std::time::SystemTime;
+use std::time::{Duration, SystemTime};
 
 pub(super) struct ObservableRetryPolicy {
     inner: ExponentialBackoff,
+    retry_window: Option<Duration>,
     metrics: Arc<HTTPClientMetrics>,
 }
 
 impl ObservableRetryPolicy {
-    pub(super) fn new(inner: ExponentialBackoff, metrics: Arc<HTTPClientMetrics>) -> Self {
-        Self { inner, metrics }
+    pub(super) fn new(
+        inner: ExponentialBackoff,
+        retry_window: Option<Duration>,
+        metrics: Arc<HTTPClientMetrics>,
+    ) -> Self {
+        Self {
+            inner,
+            retry_window,
+            metrics,
+        }
     }
 }
 
 impl RetryPolicy for ObservableRetryPolicy {
     fn should_retry(&self, request_start_time: SystemTime, n_past_retries: u32) -> RetryDecision {
-        let decision = self.inner.should_retry(request_start_time, n_past_retries);
+        let window_spent = self.retry_window.is_some_and(|window| {
+            request_start_time
+                .elapsed()
+                .is_ok_and(|elapsed| elapsed >= window)
+        });
+        let decision = if window_spent {
+            RetryDecision::DoNotRetry
+        } else {
+            self.inner.should_retry(request_start_time, n_past_retries)
+        };
         match decision {
             RetryDecision::Retry { .. } => self.metrics.record_retry(),
             RetryDecision::DoNotRetry => self.metrics.record_retries_exhausted(),
@@ -263,7 +281,7 @@ mod tests {
         let backoff = ExponentialBackoff::builder()
             .retry_bounds(Duration::from_millis(1), Duration::from_millis(2))
             .build_with_max_retries(1);
-        let policy = ObservableRetryPolicy::new(backoff, metrics.http_client());
+        let policy = ObservableRetryPolicy::new(backoff, None, metrics.http_client());
         let started = SystemTime::now();
         assert!(matches!(
             policy.should_retry(started, 0),
@@ -329,4 +347,38 @@ mod tests {
             Some(Retryable::Transient)
         ));
     }
+
+    fn policy(window: Option<Duration>) -> ObservableRetryPolicy {
+        ObservableRetryPolicy::new(
+            ExponentialBackoff::builder().build_with_max_retries(2),
+            window,
+            Arc::new(HTTPClientMetrics::new()),
+        )
+    }
+
+    #[test]
+    fn a_request_past_its_retry_window_is_not_retried() {
+        let started = SystemTime::now() - Duration::from_secs(20);
+        assert!(matches!(
+            policy(Some(Duration::from_secs(15))).should_retry(started, 0),
+            RetryDecision::DoNotRetry
+        ));
+    }
+
+    #[test]
+    fn a_request_inside_its_retry_window_is_retried() {
+        assert!(matches!(
+            policy(Some(Duration::from_secs(15))).should_retry(SystemTime::now(), 0),
+            RetryDecision::Retry { .. }
+        ));
+    }
+
+    #[test]
+    fn a_policy_without_a_window_keeps_retrying_slow_requests() {
+        let started = SystemTime::now() - Duration::from_secs(60);
+        assert!(matches!(
+            policy(None).should_retry(started, 0),
+            RetryDecision::Retry { .. }
+        ));
+    }
 }

fluxer_media_proxy/src/media_process/animated_transform.rs

@@ -182,7 +182,7 @@ impl GifResizeDims {
         (self.width as usize)
             .saturating_mul(self.height as usize)
             .saturating_mul(frames.max(1) as usize)
-            <= constants::MAX_GIF_RESIZE_OUTPUT_PIXELS
+            <= constants::MAX_ANIMATED_OUTPUT_PIXELS
     }
 }
 

fluxer_media_proxy/src/media_process/transform.rs

@@ -13,8 +13,8 @@ use super::image_probe::{
 };
 use super::loaded_image::{
     normalize_vips_image_to_uchar, page_height, resize_loaded_image, resize_loaded_image_by_scale,
-    validate_dimensions_u32, validate_jxl_decode_pixels, validate_static_output_pixels,
-    validate_vips_image,
+    static_output_pixels, validate_dimensions_u32, validate_jxl_decode_pixels,
+    validate_static_output_pixels, validate_vips_image,
 };
 use super::native_runtime::{clear_vips_error, ensure_vips_init, last_vips_error};
 use super::transform_plan::{
@@ -128,6 +128,31 @@ fn ensure_decode_cost_admissible(
     admitted
 }
 
+fn ensure_animated_output_admissible(
+    input: &[u8],
+    options: &ImageOptions,
+    media_limits: &MediaLimits,
+) -> Result<(), MediaError> {
+    let dims = probe_image_dims(media_limits, input)?;
+    if dims.pages <= 1 {
+        return Ok(());
+    }
+    let output_pixels = static_output_pixels(options, dims).saturating_mul(dims.pages as usize);
+    if output_pixels <= constants::MAX_ANIMATED_OUTPUT_PIXELS {
+        return Ok(());
+    }
+    tracing::warn!(
+        input_bytes = input.len(),
+        width = dims.width,
+        height = dims.height,
+        pages = dims.pages,
+        out_width = options.width.unwrap_or(0),
+        out_height = options.height.unwrap_or(0),
+        "animated transform refused by the output work budget"
+    );
+    Err(MediaError::TooLargeToProcess)
+}
+
 pub fn transform_image(
     input: &[u8],
     options: &ImageOptions,
@@ -184,6 +209,9 @@ pub fn transform_image(
             content_type: "image/gif",
         });
     }
+    if animated {
+        ensure_animated_output_admissible(input, options, media_limits)?;
+    }
     if should_transform_animated_webp_direct(sniffed, options, format) {
         validate_dimensions_u32(media_limits, sniffed.width, sniffed.height)?;
         let resolved = resolve_auto_quality(

fluxer_media_proxy/src/server/state.rs

@@ -42,7 +42,7 @@ impl AppState {
             NonZeroU64::new(cfg.socket_io_timeout_ms).unwrap_or(NonZeroU64::MIN);
         let store = Store::try_new(cfg.clone(), metrics.storage(), metrics.http_client())?;
         let external_client = http_client::build(
-            HTTPClientOptions::new(socket_io_timeout, socket_io_timeout).restrict_to_public(),
+            HTTPClientOptions::external(socket_io_timeout),
             metrics.http_client(),
         )?;
         let nsfw = NSFWClient::new(

fluxer_media_proxy/src/server/transform/mod.rs

@@ -574,8 +574,8 @@ async fn serve_image_transform(
             requested_download: transform.requested_download,
         });
     }
-    let can_serve_original = transform.width.is_none()
-        && transform.height.is_none()
+    let can_serve_original = ((transform.width.is_none() && transform.height.is_none())
+        || options.is_animated())
         && is_browser_displayable_image(sniffed.mime);
     let mut too_large_original = None;
     let leader_original = &mut too_large_original;

fluxer_media_proxy/src/server/transform/tests.rs

@@ -325,3 +325,35 @@ fn a_cache_larger_than_the_pod_takes_at_most_half_of_it() {
 fn the_budget_never_drops_below_its_floor() {
     assert_eq!(128 << 20, native_memory_budget_bytes(64 << 20, 0));
 }
+
+#[tokio::test]
+async fn an_animated_transform_over_the_work_budget_serves_the_original() {
+    let webp = crate::test_fixtures::ANIMATED_WEBP_1024_17_FRAMES;
+    let response = attachment_transform(
+        webp,
+        "anim.webp",
+        &[("format", "webp"), ("animated", "true"), ("width", "1000")],
+    )
+    .await;
+    assert_eq!(StatusCode::OK, response.status());
+    let body = to_bytes(response.into_body(), usize::MAX)
+        .await
+        .expect("body");
+    assert_eq!(webp, body.as_ref());
+}
+
+#[tokio::test]
+async fn an_animated_transform_within_the_work_budget_is_transformed() {
+    let webp = crate::test_fixtures::ANIMATED_WEBP_1024_17_FRAMES;
+    let response = attachment_transform(
+        webp,
+        "anim.webp",
+        &[("format", "webp"), ("animated", "true"), ("width", "128")],
+    )
+    .await;
+    assert_eq!(StatusCode::OK, response.status());
+    let body = to_bytes(response.into_body(), usize::MAX)
+        .await
+        .expect("body");
+    assert_ne!(webp, body.as_ref());
+}

fluxer_media_proxy/src/test_fixtures/images.rs

@@ -32,6 +32,7 @@ pub fn apng_header(frames: u32) -> Vec<u8> {
 
 pub const JXL_4100_LOSSY_ALPHA: &[u8] = include_bytes!("jxl_4100_lossy_alpha.jxl");
 pub const JXL_64_LOSSY_ALPHA: &[u8] = include_bytes!("jxl_64_lossy_alpha.jxl");
+pub const ANIMATED_WEBP_1024_17_FRAMES: &[u8] = include_bytes!("animated_webp_1024_17_frames.webp");
 pub const JPEG_ROTATED_MISMATCHED_SCAN: &[u8] = include_bytes!("jpeg_rotated_mismatched_scan.jpg");
 
 pub fn synthetic_png(width: u32, height: u32) -> Vec<u8> {

fluxer_media_proxy/src/test_fixtures/mod.rs

@@ -10,10 +10,11 @@ pub use adversarial::{
 };
 pub use ffmpeg_cli::{ffmpeg_gen_media, ffmpeg_gen_mp4, ffmpeg_gen_rotated_mp4, ffmpeg_mirror_mp4};
 pub use images::{
-    JPEG_ROTATED_MISMATCHED_SCAN, JXL_64_LOSSY_ALPHA, JXL_4100_LOSSY_ALPHA, animated_gif_fixture,
-    animated_gif_frames, apng_header, first_webp_anim_frame_size, gif_frame_delays_cs,
-    gif_loop_count, minimal_gif, png_dimensions, synthetic_bmp, synthetic_png,
-    webp_animation_loop_count, webp_canvas_size, webp_chunk_payloads, webp_with_metadata_chunk,
+    ANIMATED_WEBP_1024_17_FRAMES, JPEG_ROTATED_MISMATCHED_SCAN, JXL_64_LOSSY_ALPHA,
+    JXL_4100_LOSSY_ALPHA, animated_gif_fixture, animated_gif_frames, apng_header,
+    first_webp_anim_frame_size, gif_frame_delays_cs, gif_loop_count, minimal_gif, png_dimensions,
+    synthetic_bmp, synthetic_png, webp_animation_loop_count, webp_canvas_size, webp_chunk_payloads,
+    webp_with_metadata_chunk,
 };
 pub use media::{
     fixture_audio_mp3_with_png_cover_art, fixture_audio_mp4_with_attached_picture,