mirror of
https://github.com/TheFunny/TelegramTwitterMediaBot.git
synced 2026-09-23 23:32:05 +00:00
perf: bound upload-fallback preparation process-wide
Each batch's items were prepared under their own `Semaphore::new(3)`, which is not a memory bound: 8 URL workers and 4 queue workers can each be inside a batch, so a burst could have two dozen downloads in flight at once, each buffering a whole photo before it is processed. Nothing else on the media path bounds them — the send itself is paced by the rate limiter, but the download and the decode happen before it is charged. One process-wide `PREP_SLOTS` (6) replaces the per-batch semaphore, and the photo download gets its own cap: `MAX_PHOTO_DOWNLOAD_BYTES` (32 MiB) for the transfer, with `MAX_DECODE_BYTES` (512 MiB) left as the pre-allocation guard on a single decoded buffer. A photo over the download cap degrades to its smaller URL exactly as one over the decode budget does (`FallbackError::MediaTooLarge` → `fallback_url`) — never an error. Verified with the same throwaway proxy harness: 4 concurrent 10-item batches against a server that holds every response 150 ms peak at exactly 6 concurrent downloads (the per-batch three allowed 12) with all 40 items prepared. `cargo fmt --check`, `cargo clippy --workspace --all-targets --locked -- -D warnings` and `cargo test --workspace --locked` clean.
This commit is contained in:
@@ -26,9 +26,16 @@ pub const PHOTO_TARGET_DIMENSION_SUM: u32 = 9900;
|
||||
/// to a smaller media URL instead.
|
||||
pub const MAX_UPLOAD_BYTES: u64 = 10 * 1024 * 1024;
|
||||
/// Decode budget (bytes): a larger intermediate buffer is not worth the peak
|
||||
/// memory; the photo degrades to the smaller URL instead. Also the cap for
|
||||
/// downloading photos in the send fallback (they must be downloaded whole).
|
||||
/// memory; the photo degrades to the smaller URL instead.
|
||||
pub(crate) const MAX_DECODE_BYTES: u64 = 512 * 1024 * 1024;
|
||||
/// Cap for *downloading* a photo in the send fallback, kept separate from the
|
||||
/// decode budget above: the whole body is buffered before it is processed, once
|
||||
/// per download slot in flight, while the decode budget is about a single
|
||||
/// buffer. Telegram's upload cap is 10 MiB, so a photo this large can only be
|
||||
/// sent after a downscale that its reduced variant serves just as well — over
|
||||
/// the cap the item degrades to the smaller URL
|
||||
/// (`FallbackError::MediaTooLarge`), it is never an error.
|
||||
pub(crate) const MAX_PHOTO_DOWNLOAD_BYTES: u64 = 32 * 1024 * 1024;
|
||||
/// JPEG output quality (1-100).
|
||||
const JPEG_QUALITY: u8 = 90;
|
||||
|
||||
|
||||
@@ -6,11 +6,23 @@ use super::input_media::{animation_media, input_file_for, item_url, photo_media,
|
||||
use super::{MediaItemPayload, SendError, Task, classify_to_send_error, retry_delay_seconds};
|
||||
use crate::media_sender::MediaSender;
|
||||
use crate::photo::{self, MAX_UPLOAD_BYTES, PhotoPrep};
|
||||
use std::sync::LazyLock;
|
||||
use teloxide::prelude::*;
|
||||
use teloxide::types::{ChatId, InputFile, InputMedia, MessageId};
|
||||
use tempfile::NamedTempFile;
|
||||
use x_media::site::FetchError;
|
||||
|
||||
/// How many fallback items may be downloaded and processed at once, across the
|
||||
/// whole process. A per-batch bound is not a memory bound: `URL_WORKERS` (8)
|
||||
/// and the queue's workers (4) can each be inside a batch, so a per-batch three
|
||||
/// allowed two dozen downloads in flight, each buffering a whole photo
|
||||
/// (up to [`photo::MAX_PHOTO_DOWNLOAD_BYTES`]) before it is processed. This is
|
||||
/// the only admission control on the media path; the send itself is paced by
|
||||
/// the rate limiter.
|
||||
const PREP_CONCURRENCY: usize = 6;
|
||||
static PREP_SLOTS: LazyLock<tokio::sync::Semaphore> =
|
||||
LazyLock::new(|| tokio::sync::Semaphore::new(PREP_CONCURRENCY));
|
||||
|
||||
/// Infers a file extension from magic bytes so Telegram detects the mime type
|
||||
/// on multipart uploads.
|
||||
pub(super) fn sniff_ext(bytes: &[u8]) -> &'static str {
|
||||
@@ -62,12 +74,13 @@ async fn download_to_temp(
|
||||
| MediaItemPayload::Animation { media, .. } => media,
|
||||
};
|
||||
// Photos are downloaded even over the upload cap so `prepare_photo` can
|
||||
// downscale / transcode them (cap = decode budget); videos and animations
|
||||
// are refused as soon as the declared size crosses the upload cap — the
|
||||
// boundary the size probe this replaced drew: a file of exactly the cap is
|
||||
// admitted (`len > max_bytes` is false), one byte over is not.
|
||||
// downscale / transcode them, up to their own download cap; videos and
|
||||
// animations are refused as soon as the declared size crosses the upload
|
||||
// cap. The limit is that cap, not `cap + 1`: a file of exactly the cap is
|
||||
// admitted (`len > max_bytes` is false), and one byte over is not — the
|
||||
// same boundary the size probe this replaced drew.
|
||||
let limit = if matches!(item, MediaItemPayload::Photo { .. }) {
|
||||
photo::MAX_DECODE_BYTES
|
||||
photo::MAX_PHOTO_DOWNLOAD_BYTES
|
||||
} else {
|
||||
MAX_UPLOAD_BYTES
|
||||
};
|
||||
@@ -276,10 +289,12 @@ pub(super) async fn prepare_upload_item(
|
||||
|
||||
/// Download-and-reupload fallback for one media batch. Files over the upload
|
||||
/// cap are not downloaded/uploaded; the item falls back to its smaller URL
|
||||
/// (which Telegram fetches itself). Items are prepared concurrently (bounded)
|
||||
/// because the downloads are network-bound; the batch is then uploaded in its
|
||||
/// original order. Returns the fallback-error without the task attached;
|
||||
/// callers wrap it with the updated task state.
|
||||
/// (which Telegram fetches itself). Items are prepared concurrently because the
|
||||
/// downloads are network-bound, under one process-wide bound ([`PREP_SLOTS`] —
|
||||
/// the URL and queue workers can each be inside a batch, so a per-batch bound
|
||||
/// would multiply); the batch is then uploaded in its original order. Returns
|
||||
/// the fallback-error without the task attached; callers wrap it with the
|
||||
/// updated task state.
|
||||
pub(super) async fn send_batch_via_upload(
|
||||
sender: &dyn MediaSender,
|
||||
chat_id: i64,
|
||||
@@ -288,7 +303,6 @@ pub(super) async fn send_batch_via_upload(
|
||||
caption: Option<&str>,
|
||||
task: Task,
|
||||
) -> Result<Vec<Message>, SendError> {
|
||||
let sem = std::sync::Arc::new(tokio::sync::Semaphore::new(3));
|
||||
let mut set = tokio::task::JoinSet::new();
|
||||
for (i, item) in batch.iter().enumerate() {
|
||||
let item_caption = if i == 0 {
|
||||
@@ -297,9 +311,8 @@ pub(super) async fn send_batch_via_upload(
|
||||
None
|
||||
};
|
||||
let item = item.clone();
|
||||
let sem = std::sync::Arc::clone(&sem);
|
||||
set.spawn(async move {
|
||||
let _permit = sem.acquire().await.expect("upload semaphore closed");
|
||||
let _permit = PREP_SLOTS.acquire().await.expect("upload semaphore closed");
|
||||
prepare_upload_item(item, i, item_caption.as_deref()).await
|
||||
});
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user