fix(api): show premium badge for premium override (#3395)

fluxer_api/src/api/user/UserHelpers.test.ts

@@ -3,6 +3,7 @@
 import {
 	canOwnerRunBots,
 	checkIsPremium,
+	getBadgePremiumType,
 	getEffectivePremiumUntil,
 	getPremiumPaymentRecoveryGraceMs,
 	isSignInRefused,
@@ -61,6 +62,31 @@ describe('checkIsPremium', () => {
 	});
 });
 
+describe('getBadgePremiumType', () => {
+	it('shows a subscription badge for a backend premium override', () => {
+		expect(getBadgePremiumType({premiumType: null, premiumFlags: PremiumFlags.ENABLED_OVERRIDE})).toBe(
+			UserPremiumTypes.SUBSCRIPTION,
+		);
+		expect(getBadgePremiumType({premiumType: UserPremiumTypes.NONE, premiumFlags: PremiumFlags.ENABLED_OVERRIDE})).toBe(
+			UserPremiumTypes.SUBSCRIPTION,
+		);
+	});
+	it('keeps the stored premium type', () => {
+		expect(
+			getBadgePremiumType({premiumType: UserPremiumTypes.LIFETIME, premiumFlags: PremiumFlags.ENABLED_OVERRIDE}),
+		).toBe(UserPremiumTypes.LIFETIME);
+	});
+	it('shows no badge without premium or while perks are disabled', () => {
+		expect(getBadgePremiumType({premiumType: null, premiumFlags: 0})).toBeUndefined();
+		expect(
+			getBadgePremiumType({
+				premiumType: null,
+				premiumFlags: PremiumFlags.ENABLED_OVERRIDE | PremiumFlags.PERKS_DISABLED,
+			}),
+		).toBeUndefined();
+	});
+});
+
 describe('account standing', () => {
 	const hour = 3_600_000;
 	function standing(flags: bigint, tempBannedUntil: Date | null = null, deletionStartedAt: Date | null = null) {

fluxer_api/src/api/user/UserHelpers.ts

@@ -9,6 +9,8 @@ import {
 	PREMIUM_PAYMENT_RECOVERY_GRACE_DAYS,
 	PremiumFlags,
 	UserFlags,
+	type UserPremiumType,
+	UserPremiumTypes,
 } from '@fluxer/constants/src/UserConstants';
 import {MS_PER_DAY} from '@fluxer/date_utils/src/DateConstants';
 import {snowflakeToDate} from '@fluxer/snowflake/src/Snowflake';
@@ -121,6 +123,21 @@ export function checkIsPremium(user: PremiumCheckable): boolean {
 	return false;
 }
 
+export function getBadgePremiumType(
+	user: Pick<PremiumCheckable, 'premiumType' | 'premiumFlags'>,
+): UserPremiumType | undefined {
+	if (user.premiumType != null && user.premiumType !== UserPremiumTypes.NONE) {
+		return user.premiumType as UserPremiumType;
+	}
+	if (
+		(user.premiumFlags & (PremiumFlags.ENABLED_OVERRIDE | PremiumFlags.PERKS_DISABLED)) ===
+		PremiumFlags.ENABLED_OVERRIDE
+	) {
+		return UserPremiumTypes.SUBSCRIPTION;
+	}
+	return undefined;
+}
+
 const PREMIUM_CLEAR_FIELDS = [
 	'premium_type',
 	'premium_since',

fluxer_api/src/api/user/services/UserAccountLookupService.ts

@@ -11,6 +11,7 @@ import type {GuildMember} from '@app/api/models/GuildMember';
 import type {User} from '@app/api/models/User';
 import type {UserRepository} from '@app/api/user/repositories/UserRepository';
 import {isUsernameTaken} from '@app/api/user/UniqueUsernames';
+import {getBadgePremiumType} from '@app/api/user/UserHelpers';
 import {hasFixedDiscriminator} from '@app/api/user/UserTag';
 import {ChannelTypes} from '@fluxer/constants/src/ChannelConstants';
 import {
@@ -100,7 +101,7 @@ export class UserAccountLookupService {
 				}
 			}
 		}
-		let premiumType = user.premiumType ?? undefined;
+		let premiumType = getBadgePremiumType(user);
 		let premiumSince = user.premiumSince ?? undefined;
 		let premiumLifetimeSequence = user.premiumLifetimeSequence ?? undefined;
 		if (restrictProfile || user.premiumFlags & PremiumFlags.BADGE_HIDDEN) {

fluxer_app/src/app/App.tsx

@@ -22,6 +22,7 @@ import {useTabKeyFocusGuard} from '@app/features/app/hooks/useTabKeyFocusGuard';
 import {type LayoutVariant, LayoutVariantProvider} from '@app/features/app/state/LayoutVariantContext';
 import RuntimeConfig from '@app/features/app/state/RuntimeConfig';
 import RuntimeCrash from '@app/features/app/state/RuntimeCrash';
+import DeveloperOptions from '@app/features/devtools/state/DeveloperOptions';
 import {showMyselfTypingHelper} from '@app/features/devtools/utils/ShowMyselfTypingHelper';
 import GatewayConnection from '@app/features/gateway/transport/GatewayConnection';
 import GatewaySessions from '@app/features/gateway/transport/GatewaySessionPool';
@@ -83,6 +84,14 @@ interface AppWrapperProps {
 	children: ReactNode;
 }
 
+function getInstanceThemeColor(): string | null {
+	const snapshot = RuntimeConfig.getSnapshotOrNull();
+	if (snapshot === null || !(DeveloperOptions.selfHostedModeOverride || snapshot.features.self_hosted)) {
+		return null;
+	}
+	return snapshot.appPublic.branding.theme_color ?? null;
+}
+
 export const AppWrapper = observer(({children}: AppWrapperProps) => {
 	const {i18n} = useLingui();
 	const reducedMotion = Accessibility.useReducedMotion;
@@ -191,9 +200,7 @@ export const AppWrapper = observer(({children}: AppWrapperProps) => {
 		messageGutter: Accessibility.messageGutter,
 		messageGroupSpacing: Accessibility.getMessageGroupSpacingValue(messageDisplayCompact),
 		hdrDisplayMode: Accessibility.hdrDisplayMode,
-		instanceThemeColor: RuntimeConfig.isSelfHosted()
-			? (RuntimeConfig.getSnapshotOrNull()?.appPublic.branding.theme_color ?? null)
-			: null,
+		instanceThemeColor: getInstanceThemeColor(),
 	});
 	useCustomThemeStyle({
 		enabledThemeCss: ThemeLibrary.activeThemeCss,

fluxer_app/src/features/accessibility/state/Accessibility.ts

@@ -784,9 +784,6 @@ class Accessibility {
 			persistLocalMotionSettings(this.localMotionSettingsSnapshot);
 		}
 		this.customThemeCss = readLocalCustomThemeCss();
-		if (this.customThemeCss !== null) {
-			persistLocalCustomThemeCss(this.customThemeCss);
-		}
 		this.customThemeCssSyncAcrossDevices = readLocalCustomThemeCssSyncAcrossDevices();
 		this.showNeko = readAndMigrateLocalShowNeko();
 		this.keepNekoStill = readAndMigrateLocalKeepNekoStill();
@@ -798,6 +795,7 @@ class Accessibility {
 		this.initializeShowNekoSessionSync();
 		this.initializeShowNekoStorageSync();
 		this.initializeVideoSeekPreviewThumbnailsStorageSync();
+		this.initializeCustomThemeCssStorageSync();
 		this.applyStartupPresentationSettings();
 		initializeStore(this, () => this.initPersistence());
 	}
@@ -1021,7 +1019,6 @@ class Accessibility {
 			},
 		});
 		await this.applyStoredZoom();
-		this.applyStoredCustomThemeCss();
 	}
 
 	private initializeMotionDetection() {
@@ -1498,12 +1495,23 @@ class Accessibility {
 		await this.applyZoom(zoomLevel);
 	}
 
-	applyStoredCustomThemeCss(): void {
-		const customThemeCss = readLocalCustomThemeCss();
-		if (customThemeCss !== null) {
-			this.customThemeCss = customThemeCss;
-			persistLocalCustomThemeCss(customThemeCss);
-		}
+	private initializeCustomThemeCssStorageSync(): void {
+		AppStorage.subscribe(
+			(event) => {
+				if (
+					event.key !== null &&
+					event.key !== ACCESSIBILITY_CUSTOM_THEME_STORAGE_KEY &&
+					event.key !== ACCESSIBILITY_CUSTOM_THEME_SYNC_STORAGE_KEY
+				) {
+					return;
+				}
+				runInAction(() => {
+					this.customThemeCss = readLocalCustomThemeCss();
+					this.customThemeCssSyncAcrossDevices = readLocalCustomThemeCssSyncAcrossDevices();
+				});
+			},
+			{source: 'external'},
+		);
 	}
 
 	private applyStartupPresentationSettings(): void {

fluxer_media_proxy/src/coalescer/mod.rs

@@ -6,13 +6,20 @@ use crate::{
     metrics::cache::CoalescerMetrics,
 };
 use parking_lot::Mutex;
-use std::{collections::HashMap, future::Future, sync::Arc, time::Instant};
+use std::{
+    collections::HashMap,
+    future::Future,
+    sync::Arc,
+    time::{Duration, Instant},
+};
 use thiserror::Error;
 use tokio::sync::{Notify, OwnedSemaphorePermit, Semaphore, TryAcquireError};
 
 #[cfg(test)]
 mod tests;
 
+const TIMED_OUT_RETRY_AFTER: Duration = Duration::from_secs(60);
+
 #[derive(Clone, Copy, Debug, Error, Eq, PartialEq)]
 pub enum CoalescerError {
     #[error("native transform queue is full")]
@@ -23,6 +30,8 @@ pub enum CoalescerError {
     BufferBudgetExhausted,
     #[error("native transform allocation failed")]
     AllocationFailed,
+    #[error("media too large to process")]
+    TooLargeToProcess,
     #[error("request timed out")]
     RequestTimeout,
     #[error("coalesced work failed")]
@@ -43,6 +52,12 @@ struct Slot {
     notify: Notify,
 }
 
+#[derive(Debug, Default)]
+struct Keys {
+    in_flight: HashMap<String, Arc<Slot>>,
+    timed_out: HashMap<String, Instant>,
+}
+
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 enum CoordinatorState {
     Active,
@@ -58,8 +73,12 @@ struct CoordinatorGuard<'a> {
 }
 
 impl CoordinatorGuard<'_> {
-    fn publish<C>(&mut self, result: Result<BudgetedBytes, CoalescerError>, complete: C)
-    where
+    fn publish<C>(
+        &mut self,
+        result: Result<BudgetedBytes, CoalescerError>,
+        timed_out: bool,
+        complete: C,
+    ) where
         C: FnOnce(&BudgetedBytes),
     {
         if let Ok(bytes) = &result {
@@ -70,7 +89,7 @@ impl CoordinatorGuard<'_> {
             assert!(state.is_none());
             *state = Some(result);
         }
-        self.coalescer.release(self.key);
+        self.coalescer.release(self.key, timed_out);
         self.state = CoordinatorState::Published;
         self.slot.notify.notify_waiters();
     }
@@ -87,14 +106,14 @@ impl Drop for CoordinatorGuard<'_> {
                 *state = Some(Err(CoalescerError::WorkCancelled));
             }
         }
-        self.coalescer.release(self.key);
+        self.coalescer.release(self.key, false);
         self.slot.notify.notify_waiters();
     }
 }
 
 #[derive(Debug)]
 pub struct ByteCoalescer {
-    in_flight: Mutex<HashMap<String, Arc<Slot>>>,
+    keys: Mutex<Keys>,
     budget: ByteBudget,
     coordinator_capacity: Arc<Semaphore>,
     waiter_capacity: Arc<Semaphore>,
@@ -114,7 +133,7 @@ impl ByteCoalescer {
         assert!(max_in_flight > 0);
         assert!(max_waiters > 0);
         Self {
-            in_flight: Mutex::new(HashMap::new()),
+            keys: Mutex::new(Keys::default()),
             budget,
             coordinator_capacity: Arc::new(Semaphore::new(max_in_flight)),
             waiter_capacity: Arc::new(Semaphore::new(max_waiters)),
@@ -134,7 +153,7 @@ impl ByteCoalescer {
             let idle = self.idle.notified();
             tokio::pin!(idle);
             idle.as_mut().enable();
-            if self.in_flight.lock().is_empty() {
+            if self.keys.lock().in_flight.is_empty() {
                 return;
             }
             idle.await;
@@ -166,11 +185,15 @@ impl ByteCoalescer {
                 _permit: permit,
                 state: CoordinatorState::Active,
             };
-            let result = work().await.map_err(coalesced_work_error).and_then(|data| {
+            let outcome = work().await;
+            let timed_out = outcome.as_ref().is_err_and(|error| {
+                error.downcast_ref::<MediaError>() == Some(&MediaError::RequestTimeout)
+            });
+            let result = outcome.map_err(coalesced_work_error).and_then(|data| {
                 data.try_into_budgeted(&self.budget)
                     .ok_or(CoalescerError::BufferBudgetExhausted)
             });
-            guard.publish(result.clone(), complete);
+            guard.publish(result.clone(), timed_out, complete);
             return result;
         }
 
@@ -201,8 +224,8 @@ impl ByteCoalescer {
     }
 
     fn claim(&self, key: &str) -> Result<SlotClaim, CoalescerError> {
-        let mut in_flight = self.in_flight.lock();
-        if let Some(existing) = in_flight.get(key) {
+        let mut keys = self.keys.lock();
+        if let Some(existing) = keys.in_flight.get(key) {
             let permit = self
                 .waiter_capacity
                 .clone()
@@ -216,6 +239,13 @@ impl ByteCoalescer {
                 })?;
             return Ok((existing.clone(), None, Some(permit)));
         }
+        if keys
+            .timed_out
+            .get(key)
+            .is_some_and(|until| Instant::now() < *until)
+        {
+            return Err(CoalescerError::RequestTimeout);
+        }
         let permit = self
             .coordinator_capacity
             .clone()
@@ -228,16 +258,22 @@ impl ByteCoalescer {
             state: Mutex::new(None),
             notify: Notify::new(),
         });
-        in_flight.insert(key.to_owned(), slot.clone());
+        keys.in_flight.insert(key.to_owned(), slot.clone());
         Ok((slot, Some(permit), None))
     }
 
-    fn release(&self, key: &str) {
+    fn release(&self, key: &str, timed_out: bool) {
         let idle = {
-            let mut in_flight = self.in_flight.lock();
-            let removed = in_flight.remove(key);
+            let mut keys = self.keys.lock();
+            let removed = keys.in_flight.remove(key);
             assert!(removed.is_some());
-            in_flight.is_empty()
+            if timed_out {
+                let now = Instant::now();
+                keys.timed_out.retain(|_, until| *until > now);
+                keys.timed_out
+                    .insert(key.to_owned(), now + TIMED_OUT_RETRY_AFTER);
+            }
+            keys.in_flight.is_empty()
         };
         if idle {
             self.idle.notify_waiters();
@@ -249,8 +285,11 @@ fn coalesced_work_error(error: anyhow::Error) -> CoalescerError {
     if let Some(known) = error.downcast_ref::<CoalescerError>().copied() {
         return known;
     }
-    if error.downcast_ref::<MediaError>() == Some(&MediaError::AllocationFailed) {
-        return CoalescerError::AllocationFailed;
+    match error.downcast_ref::<MediaError>() {
+        Some(MediaError::AllocationFailed) => return CoalescerError::AllocationFailed,
+        Some(MediaError::TooLargeToProcess) => return CoalescerError::TooLargeToProcess,
+        Some(MediaError::RequestTimeout) => return CoalescerError::RequestTimeout,
+        _ => {}
     }
     tracing::error!(error = ?error, "coalesced work failed");
     CoalescerError::WorkFailed

fluxer_media_proxy/src/coalescer/tests.rs

@@ -49,6 +49,78 @@ async fn failure_propagates() {
     assert_eq!(CoalescerError::WorkFailed, err);
 }
 
+#[tokio::test]
+async fn work_that_ran_out_its_deadline_answers_retries_without_running_again() {
+    let coalescer = coalescer();
+    let runs = AtomicU32::new(0);
+    for _ in 0..2 {
+        let err = coalescer
+            .run_once_until(
+                "slow-gif",
+                None,
+                || async {
+                    runs.fetch_add(1, Ordering::SeqCst);
+                    Err::<MediaBytes, _>(MediaError::RequestTimeout.into())
+                },
+                |_| {},
+            )
+            .await
+            .unwrap_err();
+        assert_eq!(CoalescerError::RequestTimeout, err);
+    }
+    assert_eq!(1, runs.load(Ordering::SeqCst));
+    coalescer
+        .run_once_until(
+            "another-gif",
+            None,
+            || async { Ok(MediaBytes::from(b"fresh".to_vec())) },
+            |_| {},
+        )
+        .await
+        .expect("other keys still run");
+}
+
+#[tokio::test]
+async fn shed_admissions_and_other_failures_are_not_remembered() {
+    let coalescer = coalescer();
+    let runs = AtomicU32::new(0);
+    for error in [
+        CoalescerError::RequestTimeout,
+        CoalescerError::RequestTimeout,
+        CoalescerError::AllocationFailed,
+    ] {
+        assert_eq!(
+            error,
+            coalescer
+                .run_once_until(
+                    "shed",
+                    None,
+                    || async {
+                        runs.fetch_add(1, Ordering::SeqCst);
+                        Err::<MediaBytes, _>(error.into())
+                    },
+                    |_| {},
+                )
+                .await
+                .unwrap_err()
+        );
+    }
+    for _ in 0..2 {
+        let _ = coalescer
+            .run_once_until(
+                "broken",
+                None,
+                || async {
+                    runs.fetch_add(1, Ordering::SeqCst);
+                    Err::<MediaBytes, _>(MediaError::MediaDecodeFailed.into())
+                },
+                |_| {},
+            )
+            .await;
+    }
+    assert_eq!(5, runs.load(Ordering::SeqCst));
+}
+
 #[tokio::test]
 async fn waiter_can_time_out_behind_a_slow_leader() {
     let coalescer = coalescer();

fluxer_media_proxy/src/constants.rs

@@ -16,6 +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 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/media_process/animated_transform.rs

@@ -7,6 +7,7 @@ use super::{
     MediaError, native_animated_frame_limit, native_buffer_to_media_bytes, native_status_error,
 };
 use crate::{
+    constants,
     image_transform::ImageOptions,
     media_limits::MediaLimits,
     metrics::transform::TransformMetrics,
@@ -176,6 +177,15 @@ pub(super) struct GifResizeDims {
     pub(super) height: c_int,
 }
 
+impl GifResizeDims {
+    pub(super) fn fits_work_budget(self, frames: c_int) -> bool {
+        (self.width as usize)
+            .saturating_mul(self.height as usize)
+            .saturating_mul(frames.max(1) as usize)
+            <= constants::MAX_GIF_RESIZE_OUTPUT_PIXELS
+    }
+}
+
 pub(super) fn gif_resize_dims(
     sniffed: mime::SniffInfo,
     options: &ImageOptions,

fluxer_media_proxy/src/media_process/encoding.rs

@@ -156,6 +156,9 @@ pub(super) fn encode_vips_image(
     {
         return encode_animated_apng(image, ph, animation, media_limits, animation_loop_count);
     }
+    let width = unsafe { native::fluxer_vips_image_get_width(image.as_ptr()) };
+    let height = unsafe { native::fluxer_vips_image_get_height(image.as_ptr()) };
+    let output_pixels = u64::try_from(width).unwrap_or(0) * u64::try_from(height).unwrap_or(0);
     if format == OutputFormat::GIF
         && let Some(frame_height) = page_height
         && frame_height > 0
@@ -164,7 +167,13 @@ pub(super) fn encode_vips_image(
             image,
             resolve_animation_loop_count(image, animation_loop_count),
         )?;
-        let suffix = output_suffix(format, quality, Some(frame_height), effort_override)?;
+        let suffix = output_suffix(
+            format,
+            quality,
+            Some(frame_height),
+            effort_override,
+            output_pixels,
+        )?;
         return write_vips_image_to_vec(
             image,
             &suffix,
@@ -172,7 +181,7 @@ pub(super) fn encode_vips_image(
             animation.deadline_unix_ms,
         );
     }
-    let suffix = output_suffix(format, quality, page_height, effort_override)?;
+    let suffix = output_suffix(format, quality, page_height, effort_override, output_pixels)?;
     write_vips_image_to_vec(
         image,
         &suffix,

fluxer_media_proxy/src/media_process/loaded_image.rs

@@ -67,7 +67,7 @@ pub(crate) fn validate_jxl_decode_pixels(
         .saturating_mul(dims.height as usize)
         .saturating_mul(dims.pages.max(1) as usize);
     if pixels > constants::MAX_JXL_DECODE_PIXELS {
-        return Err(MediaError::InvalidImageDimensions);
+        return Err(MediaError::TooLargeToProcess);
     }
     Ok(())
 }
@@ -96,7 +96,7 @@ pub(crate) fn validate_static_output_pixels(
     dims: ImageDimensions,
 ) -> Result<(), MediaError> {
     if static_output_pixels(options, dims) > constants::MAX_STATIC_OUTPUT_PIXELS {
-        return Err(MediaError::InvalidImageDimensions);
+        return Err(MediaError::TooLargeToProcess);
     }
     Ok(())
 }

fluxer_media_proxy/src/media_process/mod.rs

@@ -37,7 +37,9 @@ pub use metadata::{
     MetadataOptions, PreparedMetadata, finish_metadata, metadata_json_with_options,
     prepare_metadata,
 };
-pub use native_cost::{AV_NATIVE_COST_BYTES, av_native_cost, image_transform_cost, metadata_cost};
+pub use native_cost::{
+    AV_NATIVE_COST_BYTES, NativeCost, av_native_cost, image_transform_cost, metadata_cost,
+};
 pub use nsfw_processing::encode_static_image_for_nsfw;
 pub use transform::transform_image;
 pub use video_thumbnail::{
@@ -152,6 +154,8 @@ pub enum MediaError {
     MediaTransformFailed,
     #[error("invalid image dimensions")]
     InvalidImageDimensions,
+    #[error("media too large to process")]
+    TooLargeToProcess,
     #[error("unsupported media type")]
     UnsupportedMediaType,
     #[error("stream too long")]

fluxer_media_proxy/src/media_process/native_cost.rs

@@ -10,6 +10,12 @@ use crate::{
 };
 
 pub const AV_NATIVE_COST_BYTES: usize = 256 << 20;
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum NativeCost {
+    Bytes(usize),
+    UpToBudget(usize),
+}
 const ANIMATED_CANVAS_BUFFERS: usize = 8;
 const RGBA_BYTES_PER_PIXEL: usize = 4;
 
@@ -37,7 +43,7 @@ pub fn image_transform_cost(
     input: &[u8],
     options: &ImageOptions,
     media_limits: &MediaLimits,
-) -> Result<usize, MediaError> {
+) -> Result<NativeCost, MediaError> {
     ensure_vips_init()?;
     let sniffed_mime = mime::sniff(input).mime;
     let dims = probe_image_dims(media_limits, input)?;
@@ -50,27 +56,29 @@ pub fn image_transform_cost(
     } else {
         source_pixels * decode_bytes_per_pixel(sniffed_mime) + output_pixels * encode
     };
-    Ok(input.len().saturating_add(pixel_cost))
+    Ok(NativeCost::Bytes(input.len().saturating_add(pixel_cost)))
 }
 
-pub fn av_native_cost(input_len: usize) -> usize {
-    input_len
-        .min(constants::MAX_MEDIA_PROXY_BYTES)
-        .saturating_add(AV_NATIVE_COST_BYTES)
+pub fn av_native_cost(input_len: usize) -> NativeCost {
+    NativeCost::UpToBudget(
+        input_len
+            .min(constants::MAX_MEDIA_PROXY_BYTES)
+            .saturating_add(AV_NATIVE_COST_BYTES),
+    )
 }
 
-pub fn metadata_cost(input: &[u8], media_limits: &MediaLimits) -> Result<usize, MediaError> {
+pub fn metadata_cost(input: &[u8], media_limits: &MediaLimits) -> Result<NativeCost, MediaError> {
     let sniffed_mime = mime::sniff(input).mime;
     match mime::category(sniffed_mime) {
         Some(mime::Category::Image) => {
             ensure_vips_init()?;
             let dims = probe_image_dims(media_limits, input)?;
             let source_pixels = dims.width as usize * dims.height as usize;
-            Ok(input
-                .len()
-                .saturating_add(source_pixels * decode_bytes_per_pixel(sniffed_mime)))
+            Ok(NativeCost::Bytes(input.len().saturating_add(
+                source_pixels * decode_bytes_per_pixel(sniffed_mime),
+            )))
         }
         Some(mime::Category::Video | mime::Category::Audio) => Ok(av_native_cost(input.len())),
-        _ => Ok(input.len()),
+        _ => Ok(NativeCost::Bytes(input.len())),
     }
 }

fluxer_media_proxy/src/media_process/tests/animated_gif.rs

@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
-use super::super::animated_transform::gif_resize_dims;
+use super::super::animated_transform::{GifResizeDims, gif_resize_dims};
 use super::super::{
     AnimationLimits, AnimationMode, ImageOptions, ImageQuality, MediaError, ResizeMode,
 };
@@ -156,6 +156,27 @@ fn animated_gif_with_trailing_bytes_after_the_trailer_passes_through_byte_identi
     assert_eq!(gif, passed_through.bytes);
 }
 
+#[test]
+fn animated_gif_with_trailing_bytes_after_the_trailer_resizes() {
+    for tail in [&[0u8; 4096][..], b"trailing garbage"] {
+        let mut gif = animated_gif_frames(3, 20);
+        gif.extend_from_slice(tail);
+        let resized = transform_image(
+            &gif,
+            &ImageOptions {
+                width: Some(8),
+                format: OutputFormat::GIF,
+                animation: animated_mode(),
+                ..Default::default()
+            },
+        )
+        .expect("a gif with trailing bytes after the trailer still resizes");
+
+        assert_animated_gif(&resized.bytes, 3);
+        assert_eq!(vec![20u16; 3], gif_frame_delays_cs(&resized.bytes));
+    }
+}
+
 #[test]
 fn animated_gif_over_the_frame_cap_passes_through_byte_identically() {
     let gif = animated_gif_frames(5, 20);
@@ -398,3 +419,42 @@ fn animated_gif_two_bounds_fit_inside_without_distortion() {
     assert_eq!(240, dims.width);
     assert_eq!(180, dims.height);
 }
+
+#[test]
+fn the_gif_resize_work_budget_counts_output_pixels_across_every_frame() {
+    let thumbnail = GifResizeDims {
+        width: 480,
+        height: 270,
+    };
+    assert!(thumbnail.fits_work_budget(129));
+    assert!(!thumbnail.fits_work_budget(130));
+    assert!(!thumbnail.fits_work_budget(739));
+    let still = GifResizeDims {
+        width: 4096,
+        height: 4096,
+    };
+    assert!(still.fits_work_budget(1));
+    assert!(still.fits_work_budget(0));
+    assert!(!still.fits_work_budget(2));
+}
+
+#[test]
+fn an_animated_gif_resize_over_the_work_budget_passes_through_byte_identically() {
+    let mut gif = animated_gif_frames(2, 20);
+    gif[6..8].copy_from_slice(&4096u16.to_le_bytes());
+    gif[8..10].copy_from_slice(&4096u16.to_le_bytes());
+    let passed_through = transform_image(
+        &gif,
+        &ImageOptions {
+            width: Some(4095),
+            format: OutputFormat::GIF,
+            quality: ImageQuality::Lossless,
+            animation: animated_mode(),
+            ..Default::default()
+        },
+    )
+    .expect("a gif too costly to resize still passes through");
+
+    assert_eq!("image/gif", passed_through.content_type);
+    assert_eq!(gif, passed_through.bytes);
+}

fluxer_media_proxy/src/media_process/tests/encoding.rs

@@ -25,12 +25,45 @@ fn webp_suffix(requested: EncodeEffort) -> String {
         ResolvedImageQuality::High,
         None,
         Some(requested),
+        0,
     )
     .expect("webp output suffix")
     .into_string()
     .expect("utf8 output suffix")
 }
 
+#[test]
+fn png_compression_drops_for_outputs_above_a_megapixel() {
+    let png_suffix = |output_pixels| {
+        output_suffix(
+            OutputFormat::PNG,
+            ResolvedImageQuality::High,
+            None,
+            None,
+            output_pixels,
+        )
+        .expect("png output suffix")
+        .into_string()
+        .expect("utf8 output suffix")
+    };
+    assert_eq!(
+        ".png[strip,compression=9,filter=all]",
+        png_suffix(160 * 160)
+    );
+    assert_eq!(
+        ".png[strip,compression=9,filter=all]",
+        png_suffix(1024 * 1024)
+    );
+    assert_eq!(
+        ".png[strip,compression=6,filter=all]",
+        png_suffix(1024 * 1024 + 1)
+    );
+    assert_eq!(
+        ".png[strip,compression=6,filter=all]",
+        png_suffix(3024 * 4032)
+    );
+}
+
 #[test]
 fn animated_webp_default_effort_matches_fast_tier() {
     assert_eq!(2, ResolvedImageQuality::Low.default_effort(true));

fluxer_media_proxy/src/media_process/tests/fixtures.rs

@@ -69,6 +69,12 @@ pub(super) fn png_ihdr_dimensions(ihdr: &[u8]) -> Option<(u32, u32)> {
     ))
 }
 
+pub(super) fn png_zlib_level(bytes: &[u8]) -> Option<u8> {
+    let chunks = parse_png_chunks(bytes).ok()?;
+    let idat = chunks.iter().find(|chunk| chunk.kind == *b"IDAT")?;
+    Some(idat.data.get(1)? >> 6)
+}
+
 pub(super) fn test_media_limits() -> MediaLimits {
     MediaLimits::default_from_config()
 }

fluxer_media_proxy/src/media_process/tests/metadata.rs

@@ -3,7 +3,7 @@
 use super::super::nsfw_processing::{VideoNSFWFramesRequest, extract_video_frames_for_nsfw};
 use super::super::placeholder::{encode_thumbhash, optional_thumbhash};
 use super::super::{
-    AV_NATIVE_COST_BYTES, MediaError, MetadataOptions, NSFW_PREVIEW_MAX_DIMENSION,
+    AV_NATIVE_COST_BYTES, MediaError, MetadataOptions, NSFW_PREVIEW_MAX_DIMENSION, NativeCost,
     extract_video_thumbnail, metadata_cost, metadata_json_with_options, probe_av_metadata,
 };
 use super::fixtures::{metadata_value, test_media_limits};
@@ -267,7 +267,7 @@ fn metadata_refuses_jxl_above_the_decode_pixel_cap() {
             .await
             .unwrap_err()
         });
-    assert_eq!(MediaError::InvalidImageDimensions, err);
+    assert_eq!(MediaError::TooLargeToProcess, err);
 }
 
 #[test]
@@ -296,7 +296,7 @@ fn metadata_refuses_an_image_past_the_buffer_bound() {
 fn native_cost_of_a_video_past_the_buffer_bound_leaves_out_the_mapped_tail() {
     let mp4 = mapped_past_the_buffer_bound(&fixture_h264_mp4());
     assert_eq!(
-        constants::MAX_MEDIA_PROXY_BYTES + AV_NATIVE_COST_BYTES,
+        NativeCost::UpToBudget(constants::MAX_MEDIA_PROXY_BYTES + AV_NATIVE_COST_BYTES),
         metadata_cost(&mp4, &test_media_limits()).expect("video cost")
     );
 }

fluxer_media_proxy/src/media_process/tests/transform.rs

@@ -3,13 +3,14 @@
 use super::super::image_probe::load_image;
 use super::super::native_runtime::ensure_vips_init;
 use super::super::transform::source_supports_pages;
-use super::super::{ImageOptions, MediaError, image_transform_cost};
-use super::fixtures::{animated_mode, metadata_value, transform_image};
+use super::super::{ImageOptions, MediaError, NativeCost, image_transform_cost};
+use super::fixtures::{animated_mode, metadata_value, png_zlib_level, transform_image};
 use crate::{
     mime, native,
     output_format::OutputFormat,
     test_fixtures::{
-        JXL_64_LOSSY_ALPHA, JXL_4100_LOSSY_ALPHA, png_dimensions, synthetic_bmp, synthetic_png,
+        JPEG_ROTATED_MISMATCHED_SCAN, JXL_64_LOSSY_ALPHA, JXL_4100_LOSSY_ALPHA, png_dimensions,
+        synthetic_bmp, synthetic_png,
     },
 };
 
@@ -118,7 +119,7 @@ fn refuses_jxl_above_the_decode_pixel_cap_before_decoding() {
             },
         )
         .unwrap_err();
-        assert_eq!(MediaError::InvalidImageDimensions, err);
+        assert_eq!(MediaError::TooLargeToProcess, err);
     }
 }
 
@@ -147,7 +148,7 @@ fn refuses_full_size_svg_raster_above_the_static_output_cap() {
         },
     )
     .unwrap_err();
-    assert_eq!(MediaError::InvalidImageDimensions, err);
+    assert_eq!(MediaError::TooLargeToProcess, err);
     let err = transform_image(
         svg,
         &ImageOptions {
@@ -157,7 +158,7 @@ fn refuses_full_size_svg_raster_above_the_static_output_cap() {
         },
     )
     .unwrap_err();
-    assert_eq!(MediaError::InvalidImageDimensions, err);
+    assert_eq!(MediaError::TooLargeToProcess, err);
 }
 
 #[test]
@@ -188,6 +189,9 @@ fn transform_cost_charges_the_full_source_decode_for_jxl() {
         &super::fixtures::test_media_limits(),
     )
     .unwrap();
+    let NativeCost::Bytes(cost) = cost else {
+        panic!("an image is charged exact bytes")
+    };
     assert!(cost >= 4100 * 4100 * 40, "cost {cost}");
 }
 
@@ -205,5 +209,34 @@ fn transform_cost_charges_svg_at_its_output_size() {
         &super::fixtures::test_media_limits(),
     )
     .unwrap();
+    let NativeCost::Bytes(cost) = cost else {
+        panic!("an image is charged exact bytes")
+    };
     assert!(cost < 1 << 20, "cost {cost}");
 }
+
+#[test]
+fn transforms_a_rotated_jpeg_with_mismatched_scan_headers_at_full_size() {
+    let out = transform_image(
+        JPEG_ROTATED_MISMATCHED_SCAN,
+        &ImageOptions {
+            format: OutputFormat::PNG,
+            ..Default::default()
+        },
+    )
+    .unwrap();
+    assert_eq!(Some((768, 1024)), png_dimensions(&out.bytes));
+}
+
+#[test]
+fn png_outputs_above_a_megapixel_use_default_zlib_compression() {
+    let png_options = ImageOptions {
+        format: OutputFormat::PNG,
+        ..Default::default()
+    };
+    let small = transform_image(&synthetic_bmp(160, 160), &png_options).unwrap();
+    assert_eq!(Some(3), png_zlib_level(&small.bytes));
+    let large = transform_image(&synthetic_bmp(1100, 1000), &png_options).unwrap();
+    assert_eq!(Some((1100, 1000)), png_dimensions(&large.bytes));
+    assert_eq!(Some(2), png_zlib_level(&large.bytes));
+}

fluxer_media_proxy/src/media_process/tests/video.rs

@@ -4,12 +4,13 @@ use super::super::{
     MediaError, VideoThumbnailOptions, extract_video_thumbnail,
     extract_video_thumbnail_with_options,
 };
-use super::fixtures::{decode_rgba, metadata_value, test_media_limits};
+use super::fixtures::{decode_rgba, metadata_value, png_zlib_level, test_media_limits};
 use crate::{
     image_quality::ResolvedImageQuality,
     output_format::OutputFormat,
     test_fixtures::{
-        ffmpeg_gen_media, ffmpeg_gen_mp4, ffmpeg_gen_rotated_mp4, ffmpeg_mirror_mp4, png_dimensions,
+        ffmpeg_gen_media, ffmpeg_gen_mp4, ffmpeg_gen_rotated_mp4, ffmpeg_mirror_mp4,
+        fixture_h264_mp4, png_dimensions,
     },
 };
 
@@ -35,6 +36,14 @@ fn color_tagged_video(primaries: &str, transfer: &str, matrix: &str) -> Option<V
     )
 }
 
+#[test]
+fn a_png_video_thumbnail_uses_default_zlib_compression() {
+    let thumb =
+        extract_video_thumbnail(&fixture_h264_mp4(), OutputFormat::PNG, &test_media_limits())
+            .expect("h264 video should thumbnail");
+    assert_eq!(Some(2), png_zlib_level(&thumb.bytes));
+}
+
 #[test]
 fn video_thumbnail_accepts_smpte170m_sd_primaries() {
     let Some(video) = color_tagged_video("smpte170m", "smpte170m", "smpte170m") else {

fluxer_media_proxy/src/media_process/transform.rs

@@ -161,10 +161,23 @@ pub fn transform_image(
         && !options.wants_cover_crop()
         && sniffed.mime == "image/gif"
     {
-        let bytes = if let Some(dims) = gif_resize_dims(sniffed, options) {
-            resize_animated_gif_with_ffmpeg(input, dims, options, media_limits)?
-        } else {
-            input.to_vec()
+        let bytes = match gif_resize_dims(sniffed, options) {
+            Some(dims) => {
+                let frames = probe_animated(media_limits, input)?.map_or(1, |probe| probe.pages);
+                if dims.fits_work_budget(frames) {
+                    resize_animated_gif_with_ffmpeg(input, dims, options, media_limits)?
+                } else {
+                    tracing::info!(
+                        input_bytes = input.len(),
+                        width = dims.width,
+                        height = dims.height,
+                        frames,
+                        "animated gif resize over the work budget, serving the source"
+                    );
+                    input.to_vec()
+                }
+            }
+            None => input.to_vec(),
         };
         return Ok(ProcessedMedia {
             bytes,
@@ -382,13 +395,9 @@ pub fn transform_image(
     }
 
     let loader_options = if animated && source_supports_pages(sniffed.mime) {
-        if sniffed.mime == "image/jpeg" {
-            "n=-1,access=sequential"
-        } else {
-            "n=-1,access=sequential,fail=true"
-        }
+        "n=-1,access=sequential,fail=true"
     } else if sniffed.mime == "image/jpeg" {
-        "access=sequential"
+        ""
     } else {
         "access=sequential,fail=true"
     };

fluxer_media_proxy/src/media_process/transform_plan.rs

@@ -13,12 +13,14 @@ use std::ffi::CString;
 const AUTO_LOSSLESS_PALETTE_ANIMATION_MAX_BYTES: usize = 4 * 1024 * 1024;
 const AUTO_LOSSLESS_PALETTE_ANIMATION_MAX_PIXELS: usize = 16 * 1024 * 1024;
 const VIPS_WEBP_MAX_EFFORT: u8 = 6;
+const PNG_MAX_COMPRESSION_MAX_PIXELS: u64 = 1024 * 1024;
 
 pub(super) fn output_suffix(
     format: OutputFormat,
     quality: ResolvedImageQuality,
     page_height: Option<c_int>,
     effort_override: Option<EncodeEffort>,
+    output_pixels: u64,
 ) -> Result<CString, MediaError> {
     let q = quality.encoder_quality();
     let animated = page_height.is_some();
@@ -43,12 +45,19 @@ pub(super) fn output_suffix(
                 ),
             }
         }
-        OutputFormat::PNG | OutputFormat::APNG => match page_height {
-            Some(ph) if format == OutputFormat::APNG => {
-                format!(".png[strip,compression=9,filter=all,page-height={ph}]")
+        OutputFormat::PNG | OutputFormat::APNG => {
+            let compression = if output_pixels > PNG_MAX_COMPRESSION_MAX_PIXELS {
+                6
+            } else {
+                9
+            };
+            match page_height {
+                Some(ph) if format == OutputFormat::APNG => {
+                    format!(".png[strip,compression={compression},filter=all,page-height={ph}]")
+                }
+                _ => format!(".png[strip,compression={compression},filter=all]"),
             }
-            _ => ".png[strip,compression=9,filter=all]".to_owned(),
-        },
+        }
         OutputFormat::GIF => match page_height {
             Some(ph) => {
                 format!(".gif[strip,dither=1.0,effort=7,interframe_maxerror=8.0,page-height={ph}]")

fluxer_media_proxy/src/media_process/video_thumbnail.rs

@@ -54,7 +54,7 @@ pub fn extract_video_thumbnail_with_options(
     } = options;
     let runtime = vips_runtime()?;
     let (max_width, max_height) = native_thumbnail_bounds(width, height, media_limits)?;
-    let suffix = output_suffix(format, quality, None, None)?;
+    let suffix = output_suffix(format, quality, None, None, u64::MAX)?;
     let mut out_ptr: *mut c_void = ptr::null_mut();
     let mut out_size: size_t = 0;
     let mut out_capacity: size_t = 0;

fluxer_media_proxy/src/native_shim/gif_timing.c

@@ -193,7 +193,7 @@ int fluxer_read_gif_frame_delays(const uint8_t *bytes, size_t len,
         block_count++;
         uint8_t introducer = bytes[offset++];
         if (introducer == 0x3b) {
-            if (offset != len || has_pending_delay || frame_count != expected_frames) {
+            if (has_pending_delay || frame_count != expected_frames) {
                 return FLUXER_NATIVE_STATUS_CODEC_FAILURE;
             }
             if (out_loop_count != NULL) *out_loop_count = loop_count;

fluxer_media_proxy/src/native_shim/gif_validation.c

@@ -148,7 +148,7 @@ int fluxer_validate_complete_gif(const uint8_t *bytes, size_t len,
         context.block_count++;
         uint8_t introducer = context.bytes[context.offset++];
         if (introducer == 0x3b) {
-            if (context.image_count == 0 || context.offset != context.len ||
+            if (context.image_count == 0 ||
                 context.image_count > (size_t)INT_MAX) {
                 return FLUXER_NATIVE_STATUS_CODEC_FAILURE;
             }

fluxer_media_proxy/src/server/format_policy.rs

@@ -184,6 +184,13 @@ pub(in crate::server) fn content_type_is_trustworthy(content_type: &str) -> bool
     )
 }
 
+pub(in crate::server) fn is_browser_displayable_image(mime: &str) -> bool {
+    matches!(
+        mime,
+        "image/jpeg" | "image/png" | "image/gif" | "image/webp"
+    )
+}
+
 pub(in crate::server) fn is_svg_content_type(content_type: &str) -> bool {
     mime::normalize(Some(content_type))
         .is_some_and(|value| value.eq_ignore_ascii_case("image/svg+xml"))

fluxer_media_proxy/src/server/native_task_executor.rs

@@ -2,7 +2,7 @@
 
 use crate::{
     byte_budget::ByteBudget,
-    media_process::MediaError,
+    media_process::{MediaError, NativeCost},
     metrics::{
         self,
         transform::{NativeTransformMetrics, TransformMetrics},
@@ -96,7 +96,7 @@ impl NativeTaskExecutor {
     ) -> anyhow::Result<T>
     where
         T: Send + 'static,
-        C: FnOnce() -> anyhow::Result<usize> + Send + 'static,
+        C: FnOnce() -> anyhow::Result<NativeCost> + Send + 'static,
         F: FnOnce() -> anyhow::Result<T> + Send + 'static,
     {
         let admission = match self.native_transforms.try_admit() {
@@ -112,8 +112,10 @@ impl NativeTaskExecutor {
             .native_transforms
             .wait_until_admitted(&admission, deadline)
             .await;
+        let admitted = Instant::now();
+        let waited = admitted.duration_since(wait_started);
         self.native_metrics
-            .observe_wait(metrics::duration_millis(wait_started.elapsed()));
+            .observe_wait(metrics::duration_millis(waited));
         let permit = permit?;
         let memory = self.memory.clone();
         let capacity = self.memory_budget_bytes;
@@ -121,14 +123,17 @@ impl NativeTaskExecutor {
         self.run_task(deadline, move || {
             let _span = span.enter();
             let _permit = permit;
-            let bytes = cost()?;
-            if bytes > capacity {
-                warn!(
-                    bytes,
-                    capacity, "native work refused, larger than the memory budget"
-                );
-                return Err(MediaError::InvalidImageDimensions.into());
-            }
+            let bytes = match cost()? {
+                NativeCost::Bytes(bytes) if bytes > capacity => {
+                    warn!(
+                        bytes,
+                        capacity, "native work refused, larger than the memory budget"
+                    );
+                    return Err(MediaError::TooLargeToProcess.into());
+                }
+                NativeCost::Bytes(bytes) => bytes,
+                NativeCost::UpToBudget(bytes) => bytes.min(capacity),
+            };
             let Some(_reserved) = memory.try_reserve(bytes) else {
                 warn!(bytes, capacity, "native work refused, memory budget in use");
                 return Err(MediaError::AllocationFailed.into());
@@ -136,6 +141,15 @@ impl NativeTaskExecutor {
             work()
         })
         .await
+        .map_err(|error| {
+            let ran_short = deadline
+                .is_some_and(|deadline| deadline.saturating_duration_since(admitted) < waited);
+            if ran_short && error.downcast_ref() == Some(&MediaError::RequestTimeout) {
+                TimedSemaphoreError::RequestTimeout.into()
+            } else {
+                error
+            }
+        })
     }
 
     pub(in crate::server) fn begin_shutdown(&self) {
@@ -197,7 +211,7 @@ impl NativeTaskExecutor {
         let completion = match deadline {
             Some(deadline) => match tokio::time::timeout_at(deadline.into(), receiver).await {
                 Ok(completion) => completion,
-                Err(_) => return Err(TimedSemaphoreError::RequestTimeout.into()),
+                Err(_) => return Err(MediaError::RequestTimeout.into()),
             },
             None => receiver.await,
         };
@@ -298,7 +312,7 @@ mod tests {
         let error = executor
             .run_native(
                 Some(Instant::now() + Duration::from_millis(30)),
-                || Ok(0),
+                || Ok(NativeCost::Bytes(0)),
                 move || {
                     released.blocking_recv();
                     let _ = finished.blocking_send(());
@@ -308,8 +322,8 @@ mod tests {
             .await
             .expect_err("the task outlives its deadline");
         assert_eq!(
-            Some(&TimedSemaphoreError::RequestTimeout),
-            error.downcast_ref::<TimedSemaphoreError>()
+            Some(&MediaError::RequestTimeout),
+            error.downcast_ref::<MediaError>()
         );
         let rendered = metrics.render();
         assert_eq!(
@@ -344,6 +358,49 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn a_task_admitted_late_times_out_as_an_admission_timeout() {
+        let metrics = metrics::Metrics::new();
+        let executor = Arc::new(executor(&metrics, 1, 1));
+        let (release_holder, mut holder_released) = mpsc::channel::<()>(1);
+        let (release_queued, mut queued_released) = mpsc::channel::<()>(1);
+        let holder = Arc::clone(&executor);
+        let held = tokio::spawn(async move {
+            holder
+                .run_native(
+                    None,
+                    || Ok(NativeCost::Bytes(0)),
+                    move || {
+                        holder_released.blocking_recv();
+                        Ok(())
+                    },
+                )
+                .await
+        });
+        tokio::time::sleep(Duration::from_millis(20)).await;
+        let deadline = Instant::now() + Duration::from_millis(300);
+        let queued = executor.run_native(
+            Some(deadline),
+            || Ok(NativeCost::Bytes(0)),
+            move || {
+                queued_released.blocking_recv();
+                Ok(())
+            },
+        );
+        let release = async {
+            tokio::time::sleep(Duration::from_millis(240)).await;
+            drop(release_holder);
+        };
+        let (error, ()) = tokio::join!(queued, release);
+        let error = error.expect_err("the queued task outlives its deadline");
+        assert_eq!(
+            Some(&TimedSemaphoreError::RequestTimeout),
+            error.downcast_ref::<TimedSemaphoreError>()
+        );
+        drop(release_queued);
+        held.await.expect("held task").expect("held work");
+    }
+
     #[tokio::test]
     async fn a_task_that_completes_before_its_deadline_is_never_detached() {
         let metrics = metrics::Metrics::new();
@@ -351,7 +408,7 @@ mod tests {
         let value = executor
             .run_native(
                 Some(Instant::now() + Duration::from_secs(30)),
-                || Ok(0),
+                || Ok(NativeCost::Bytes(0)),
                 || Ok(7u32),
             )
             .await
@@ -378,7 +435,7 @@ mod tests {
             holder
                 .run_native(
                     None,
-                    || Ok(0),
+                    || Ok(NativeCost::Bytes(0)),
                     move || {
                         released.blocking_recv();
                         Ok(())
@@ -388,7 +445,7 @@ mod tests {
         });
         tokio::time::sleep(Duration::from_millis(50)).await;
         let error = executor
-            .run_native(None, || Ok(0), || Ok(()))
+            .run_native(None, || Ok(NativeCost::Bytes(0)), || Ok(()))
             .await
             .expect_err("the admission queue is full");
         assert_eq!(
@@ -412,7 +469,7 @@ mod tests {
         let executor = executor(&metrics, 2, 4);
         executor.begin_shutdown();
         let error = executor
-            .run_native(None, || Ok(0), || Ok(()))
+            .run_native(None, || Ok(NativeCost::Bytes(0)), || Ok(()))
             .await
             .expect_err("a closed executor admits nothing");
         assert_eq!(
@@ -429,13 +486,13 @@ mod tests {
         let error = executor
             .run_native(
                 None,
-                || Ok(1025),
+                || Ok(NativeCost::Bytes(1025)),
                 || -> anyhow::Result<()> { panic!("refused work must not run") },
             )
             .await
             .expect_err("the job exceeds the whole budget");
         assert_eq!(
-            Some(&MediaError::InvalidImageDimensions),
+            Some(&MediaError::TooLargeToProcess),
             error.downcast_ref::<MediaError>()
         );
     }
@@ -451,7 +508,7 @@ mod tests {
             holder
                 .run_native(
                     None,
-                    || Ok(1000),
+                    || Ok(NativeCost::Bytes(1000)),
                     move || {
                         let _ = started.blocking_send(());
                         released.blocking_recv();
@@ -462,7 +519,7 @@ mod tests {
         });
         has_started.recv().await;
         let error = executor
-            .run_native(None, || Ok(100), || Ok(()))
+            .run_native(None, || Ok(NativeCost::Bytes(100)), || Ok(()))
             .await
             .expect_err("the budget is in use");
         assert_eq!(
@@ -472,8 +529,19 @@ mod tests {
         drop(release);
         held.await.expect("held task").expect("held work");
         executor
-            .run_native(None, || Ok(1024), || Ok(()))
+            .run_native(None, || Ok(NativeCost::Bytes(1024)), || Ok(()))
             .await
             .expect("the released budget admits a job of the full size");
     }
+
+    #[tokio::test]
+    async fn work_that_reserves_up_to_the_budget_runs_on_a_budget_smaller_than_its_estimate() {
+        let metrics = metrics::Metrics::new();
+        let executor = executor(&metrics, 1, 0);
+        let value = executor
+            .run_native(None, || Ok(NativeCost::UpToBudget(1 << 30)), || Ok(7u32))
+            .await
+            .expect("the job takes the whole budget instead of being refused");
+        assert_eq!(7, value);
+    }
 }

fluxer_media_proxy/src/server/routes/internal.rs

@@ -685,7 +685,7 @@ mod tests {
                 .tasks()
                 .run_native(
                     None,
-                    || Ok(0),
+                    || Ok(media_process::NativeCost::Bytes(0)),
                     move || {
                         let _ = started.blocking_send(());
                         let _ = released.blocking_recv();

fluxer_media_proxy/src/server/stored/mod.rs

@@ -17,7 +17,8 @@ use crate::{
         attachment_signature,
         format_policy::{
             OriginalImageRequest, default_transform_quality,
-            effective_animated_image_output_format, extension_from_mime, is_svg_content_type,
+            effective_animated_image_output_format, extension_from_mime,
+            is_browser_displayable_image, is_svg_content_type,
             same_format_loaded_image_request_can_use_original_with_sniff,
         },
         media_operations::spool_stored_media,
@@ -196,6 +197,7 @@ pub(in crate::server) async fn serve_asset_image(
                 object,
                 range_header,
                 source_format,
+                sniffed_content_type: sniffed.mime,
                 detail,
             });
         }
@@ -219,11 +221,19 @@ struct AssetTransformFailure<'a> {
     object: storage::Object,
     range_header: Option<&'a str>,
     source_format: AssetExtension,
+    sniffed_content_type: &'static str,
     detail: String,
 }
 
 fn asset_transform_failure_response(failure: AssetTransformFailure<'_>) -> Response {
-    let src_ct = failure.object.content_type.as_str();
+    let stored_ct = failure.object.content_type.as_str();
+    let src_ct = if !stored_ct.starts_with("image/")
+        && is_browser_displayable_image(failure.sniffed_content_type)
+    {
+        failure.sniffed_content_type
+    } else {
+        stored_ct
+    };
     let src_is_displayable = src_ct.starts_with("image/")
         && src_ct != "image/avif"
         && src_ct != "image/heic"
@@ -234,7 +244,7 @@ fn asset_transform_failure_response(failure: AssetTransformFailure<'_>) -> Respo
         return media_response(MediaResponse {
             method: failure.method,
             data: failure.object.data.into(),
-            content_type: &failure.object.content_type,
+            content_type: src_ct,
             range_header: failure.range_header,
             disposition: None,
         });

fluxer_media_proxy/src/server/stored/tests.rs

@@ -2,7 +2,9 @@
 
 use super::*;
 use crate::{
-    config::Config, output_format::OutputFormat, server::asset_path::parse_standard_asset_path,
+    config::Config,
+    output_format::OutputFormat,
+    server::asset_path::{parse_simple_asset_path, parse_standard_asset_path},
     storage::Object,
 };
 use axum::{body::to_bytes, http::header};
@@ -48,6 +50,7 @@ fn asset_transform_failure(error: CoalescerError) -> Response {
         object: asset_source_object(),
         range_header: None,
         source_format: AssetExtension::Png,
+        sniffed_content_type: "image/png",
         detail: "asset key=avatars/1/abc".to_owned(),
     })
 }
@@ -239,6 +242,7 @@ fn a_transform_failure_on_a_source_the_browser_cannot_show_stays_an_error() {
             },
             range_header: None,
             source_format: AssetExtension::Svg,
+            sniffed_content_type: "image/svg+xml",
             detail: "asset key=avatars/1/abc".to_owned(),
         })
     };
@@ -252,6 +256,56 @@ fn a_transform_failure_on_a_source_the_browser_cannot_show_stays_an_error() {
     );
 }
 
+#[test]
+fn an_octet_stream_asset_sniffed_as_an_image_browsers_cannot_show_stays_an_error() {
+    let response = asset_transform_failure_response(AssetTransformFailure {
+        method: Method::GET,
+        error: CoalescerError::TooLargeToProcess,
+        object: Object {
+            data: Bytes::from_static(&[0xff, 0x0a, 0x00, 0x00]),
+            content_type: "application/octet-stream".to_owned(),
+        },
+        range_header: None,
+        source_format: AssetExtension::Jxl,
+        sniffed_content_type: "image/jxl",
+        detail: "asset key=avatars/1/abc".to_owned(),
+    });
+    assert_eq!(StatusCode::INTERNAL_SERVER_ERROR, response.status());
+}
+
+#[tokio::test]
+async fn an_untransformable_asset_stored_as_octet_stream_serves_the_original_as_its_sniffed_type() {
+    let fake = crate::storage::tests::fake_s3().await;
+    let tmp = tempfile::tempdir().unwrap();
+    let mut cfg = fake.config(tmp.path());
+    cfg.media.transform_timeout_ms = 60_000;
+    let app = test_app_state(cfg);
+    let asset =
+        parse_simple_asset_path("/emojis/1554531234351165442.webp", AssetKind::Emoji).unwrap();
+    let mut corrupt_gif = crate::test_fixtures::minimal_gif();
+    corrupt_gif.truncate(corrupt_gif.len() - 6);
+    corrupt_gif.extend_from_slice(&[0x02, 0x02, 0xff, 0xff, 0x00, 0x3b]);
+    fake.put_object(
+        &format!("{}/{}", app.cfg.storage.bucket_cdn, asset.storage_key),
+        crate::storage::tests::FakeObject {
+            body: corrupt_gif.clone(),
+            content_type: Some("application/octet-stream".to_owned()),
+            ..crate::storage::tests::FakeObject::default()
+        },
+    );
+    let response =
+        serve_asset_image(&app, Method::GET, asset, &HashMap::new(), &HeaderMap::new()).await;
+    assert_eq!(StatusCode::OK, response.status());
+    assert_eq!(
+        Some("image/gif"),
+        response
+            .headers()
+            .get(header::CONTENT_TYPE)
+            .and_then(|value| value.to_str().ok())
+    );
+    assert_eq!(&corrupt_gif[..], &response_body(response).await[..]);
+}
+
 #[tokio::test]
 async fn a_stored_passthrough_stream_declares_the_length_it_serves() {
     let tmp = tempfile::tempdir().unwrap();

fluxer_media_proxy/src/server/transform/cache.rs

@@ -52,7 +52,10 @@ where
         .cache()
         .get_or_run(cache_key, format, deadline, work)
         .await;
-    if matches!(result, Err(CoalescerError::WorkFailed)) {
+    if matches!(
+        result,
+        Err(CoalescerError::WorkFailed | CoalescerError::TooLargeToProcess)
+    ) {
         runtime.metrics().record_transform_failure();
     }
     result
@@ -63,7 +66,7 @@ pub(in crate::server) fn coalescer_failure_response(
     timeout_code: &'static str,
 ) -> Option<Response> {
     let reason = match error {
-        CoalescerError::WorkFailed => return None,
+        CoalescerError::WorkFailed | CoalescerError::TooLargeToProcess => return None,
         CoalescerError::RequestTimeout => {
             return Some(text_with_reason(
                 StatusCode::GATEWAY_TIMEOUT,

fluxer_media_proxy/src/server/transform/execution.rs

@@ -123,19 +123,17 @@ pub(in crate::server) async fn run_video_transform(
 }
 
 pub(in crate::server) fn coalesced_work_result<T>(result: anyhow::Result<T>) -> anyhow::Result<T> {
-    match result {
-        Ok(value) => Ok(value),
-        Err(error) if transform_error_is_timeout(&error) => {
-            Err(anyhow::Error::new(CoalescerError::RequestTimeout))
+    let Err(error) = result else {
+        return result;
+    };
+    let retyped = match error.downcast_ref::<TimedSemaphoreError>() {
+        Some(TimedSemaphoreError::QueueFull | TimedSemaphoreError::RequestTimeout) => {
+            CoalescerError::RequestTimeout
         }
-        Err(error)
-            if error.downcast_ref::<TimedSemaphoreError>()
-                == Some(&TimedSemaphoreError::Closed) =>
-        {
-            Err(anyhow::Error::new(CoalescerError::Unavailable))
-        }
-        Err(error) => Err(error),
-    }
+        Some(TimedSemaphoreError::Closed) => CoalescerError::Unavailable,
+        None => return Err(error),
+    };
+    Err(anyhow::Error::new(retyped))
 }
 
 pub(in crate::server) fn transform_error_is_timeout(error: &anyhow::Error) -> bool {
@@ -187,6 +185,12 @@ mod tests {
             Some(&MediaError::MediaDecodeFailed),
             failed.downcast_ref::<MediaError>()
         );
+        let ran_out = coalesced_work_result::<()>(Err(MediaError::RequestTimeout.into()))
+            .expect_err("work that ran out its deadline passes through");
+        assert_eq!(
+            Some(&MediaError::RequestTimeout),
+            ran_out.downcast_ref::<MediaError>()
+        );
     }
 
     #[test]

fluxer_media_proxy/src/server/transform/mod.rs

@@ -16,7 +16,7 @@ use crate::{
     image_quality::ImageQuality,
     image_transform::{AnimationLimits, AnimationMode, ImageOptions, ResizeMode},
     media_limits::MediaLimits,
-    media_process::MediaBytes,
+    media_process::{MediaBytes, MediaError},
     metrics::{self, transform::TransformMetrics},
     mime,
     output_format::OutputFormat,
@@ -24,7 +24,8 @@ use crate::{
         format_policy::{
             OriginalImageRequest, content_type_is_trustworthy, default_transform_quality,
             effective_animated_image_output_format, extension_from_mime,
-            external_default_output_extension, image_extension_from_filename, is_svg_content_type,
+            external_default_output_extension, image_extension_from_filename,
+            is_browser_displayable_image, is_svg_content_type,
             same_format_loaded_image_request_can_use_original_with_sniff, source_image_format,
             transform_response_content_type, transform_static_quality_default,
         },
@@ -573,20 +574,42 @@ async fn serve_image_transform(
             requested_download: transform.requested_download,
         });
     }
+    let can_serve_original = transform.width.is_none()
+        && transform.height.is_none()
+        && is_browser_displayable_image(sniffed.mime);
+    let mut too_large_original = None;
+    let leader_original = &mut too_large_original;
     let transformed = match cached_transform(CachedTransformRequest {
         runtime,
         cache_key,
         format: options.format,
         deadline: deadline_instant(options.deadline_ms),
         work: move || async move {
-            coalesced_work_result(run_transform(runtime, data, options).await)
-                .map(|media| MediaBytes::from(media.bytes))
+            let original = can_serve_original.then(|| data.clone());
+            let result = coalesced_work_result(run_transform(runtime, data, options).await);
+            if result
+                .as_ref()
+                .is_err_and(|error| error.downcast_ref() == Some(&MediaError::TooLargeToProcess))
+            {
+                *leader_original = original;
+            }
+            result.map(|media| MediaBytes::from(media.bytes))
         },
     })
     .await
     {
         Ok(bytes) => bytes.as_bytes().clone(),
         Err(error) => {
+            if let Some(original) = too_large_original {
+                return original_bytes_response(OriginalBytes {
+                    method,
+                    data: original,
+                    content_type: sniffed.mime,
+                    filename,
+                    range_header,
+                    requested_download: transform.requested_download,
+                });
+            }
             if let Some(response) = coalescer_failure_response(error, "coalescer_timeout_image") {
                 return response;
             }

fluxer_media_proxy/src/server/transform/tests.rs

@@ -4,8 +4,9 @@ use super::*;
 use crate::{
     constants::AssetExtension,
     server::format_policy::{default_transform_quality, transform_static_quality_default},
-    test_fixtures::{fixture_h264_mp4, fixture_jpeg, synthetic_png},
+    test_fixtures::{JXL_4100_LOSSY_ALPHA, fixture_h264_mp4, fixture_jpeg, synthetic_png},
 };
+use axum::body::to_bytes;
 
 const TRANSFORM_WIDTH: u32 = 32;
 
@@ -200,6 +201,112 @@ async fn a_redundant_format_parameter_shares_the_implicit_transform_cache_entry(
     }
 }
 
+async fn attachment_transform(data: &[u8], filename: &str, params: &[(&str, &str)]) -> Response {
+    let params = params
+        .iter()
+        .map(|(name, value)| ((*name).to_owned(), (*value).to_owned()))
+        .collect::<HashMap<_, _>>();
+    serve_bytes_or_transform(
+        &test_runtime(),
+        ServeBytesRequest {
+            method: Method::GET,
+            data: Bytes::copy_from_slice(data),
+            content_type: "application/octet-stream".to_owned(),
+            cache_identity: filename,
+            filename,
+            route: TransformRoute::Attachment,
+            params: &params,
+            headers: &HeaderMap::new(),
+        },
+    )
+    .await
+}
+
+#[tokio::test]
+async fn a_full_size_transform_too_large_to_process_serves_the_original() {
+    let png = synthetic_png(6145, 4097);
+    let response = attachment_transform(&png, "big.png", &[("format", "webp")]).await;
+    assert_eq!(StatusCode::OK, response.status());
+    assert_eq!(
+        Some("image/png"),
+        response
+            .headers()
+            .get(header::CONTENT_TYPE)
+            .and_then(|value| value.to_str().ok())
+    );
+    let body = to_bytes(response.into_body(), usize::MAX)
+        .await
+        .expect("body");
+    assert_eq!(png, body);
+}
+
+#[tokio::test]
+async fn a_waiter_releases_its_source_while_the_leader_runs() {
+    const KEY: &str = "attachment:art.png|asset_kind=not-applicable|w=none|h=none|fmt=webp|q=lossless|anim=false|effort=none|resize=fit";
+    let runtime = test_runtime();
+    let params = HashMap::from([("format".to_owned(), "webp".to_owned())]);
+    let headers = HeaderMap::new();
+    let source = Bytes::from(synthetic_png(48, 48));
+    let probe = source.clone();
+    let (release, released) = tokio::sync::oneshot::channel::<()>();
+    let leader =
+        runtime
+            .cache()
+            .get_or_run(KEY.to_owned(), OutputFormat::WebP, None, || async move {
+                let _ = released.await;
+                Ok(MediaBytes::from(vec![0u8; 4]))
+            });
+    let waiter = async {
+        let response = serve_bytes_or_transform(
+            &runtime,
+            ServeBytesRequest {
+                method: Method::GET,
+                data: source,
+                content_type: "application/octet-stream".to_owned(),
+                cache_identity: "art.png",
+                filename: "art.png",
+                route: TransformRoute::Attachment,
+                params: &params,
+                headers: &headers,
+            },
+        );
+        tokio::pin!(response);
+        assert!(
+            tokio::time::timeout(std::time::Duration::from_millis(100), response.as_mut())
+                .await
+                .is_err(),
+            "the request waits on the running leader"
+        );
+        assert!(
+            probe.is_unique(),
+            "a waiter must not pin its source while the leader runs"
+        );
+        let _ = release.send(());
+        response.await
+    };
+    let (leader, response) = tokio::join!(leader, waiter);
+    leader.expect("leader publishes");
+    assert_eq!(StatusCode::OK, response.status());
+}
+
+#[tokio::test]
+async fn a_resized_transform_too_large_to_process_still_fails() {
+    let response = attachment_transform(
+        &synthetic_png(6145, 4097),
+        "big.png",
+        &[("format", "webp"), ("width", "6145")],
+    )
+    .await;
+    assert_eq!(StatusCode::BAD_REQUEST, response.status());
+}
+
+#[tokio::test]
+async fn a_full_size_transform_too_large_to_process_never_serves_an_undisplayable_original() {
+    let response =
+        attachment_transform(JXL_4100_LOSSY_ALPHA, "big.jxl", &[("format", "webp")]).await;
+    assert_eq!(StatusCode::BAD_REQUEST, response.status());
+}
+
 #[test]
 fn the_budget_leaves_room_for_the_cache_and_headroom() {
     assert_eq!(

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 JPEG_ROTATED_MISMATCHED_SCAN: &[u8] = include_bytes!("jpeg_rotated_mismatched_scan.jpg");
 
 pub fn synthetic_png(width: u32, height: u32) -> Vec<u8> {
     ensure_vips_init().unwrap();

fluxer_media_proxy/src/test_fixtures/mod.rs

@@ -10,10 +10,10 @@ pub use adversarial::{
 };
 pub use ffmpeg_cli::{ffmpeg_gen_media, ffmpeg_gen_mp4, ffmpeg_gen_rotated_mp4, ffmpeg_mirror_mp4};
 pub use images::{
-    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,
+    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,