Building a Video API SDK in Rust with reqwest and serde for Multi-Region Feeds

Our Asia-Pacific trending pipeline used to be one PHP cron that walked nine region codes — US, GB, JP, KR, TW, SG, VN, TH, HK — hit the upstream video API for each, paged through the results, and wrote rows into SQLite. It worked. It also took about 38 minutes of wall clock, and nearly all of that was a single process sitting on a socket waiting for a TLS handshake to finish, one request at a time. curl_multi is the answer in PHP and I have written that answer three times and enjoyed it zero times.

The fix was not rewriting the site. TopVideoHub is PHP 8.4 on LiteSpeed behind Cloudflare with SQLite FTS5 doing CJK search, and none of that is going anywhere. The fix was a small Rust binary that does exactly one job: talk to video APIs, normalize what comes back, and stream NDJSON to stdout for the PHP side to ingest. This is how that SDK layer is built — the serde modeling, the reqwest client, quota rotation across API keys, retry policy, and the seams that make it testable.

Model the response before you write a client

Every API client I have seen that started with Client::new() ended up with serde_json::Value soup at the edges and a .get("snippet").and_then(|v| v.get("title")) chain in the middle. Start from the wire format instead. Write the structs, feed them a recorded response, and only then figure out how to make requests.

Three things about this particular API shaped the model:

The envelope is generic. nextPageToken, pageInfo, and items[] are identical whether you are listing videos, channels, or categories, so the envelope is a type parameter.
Counts arrive as JSON strings. "viewCount": "1048576". In PHP that silently becomes a string and every comparison after it is a coercion you did not plan. In Rust it is a deserializer you write once.
Half the fields are conditionally absent. Live streams have no contentDetails.duration. Videos with statistics disabled have no likeCount. Option<T> and #[serde(default)] are the whole story, but you have to decide per field which one is correct — default for a count that means zero, Option for a fact you genuinely do not know.

use serde::{Deserialize, Deserializer};
use std::time::Duration;

#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct Page<T> {
    #[serde(default)]
    pub next_page_token: Option<String>,
    #[serde(default)]
    pub items: Vec<T>,
    #[serde(default)]
    pub page_info: PageInfo,
}

#[derive(Debug, Default, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PageInfo {
    pub total_results: u32,
    pub results_per_page: u32,
}

#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct Video {
    pub id: String,
    pub snippet: Snippet,
    #[serde(default)]
    pub content_details: Option<ContentDetails>,
    #[serde(default)]
    pub statistics: Option<Statistics>,
}

#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct Snippet {
    pub title: String,
    #[serde(default)]
    pub description: String,
    pub channel_id: String,
    pub channel_title: String,
    pub published_at: chrono::DateTime<chrono::Utc>,
    #[serde(default)]
    pub default_audio_language: Option<String>,
    #[serde(default)]
    pub tags: Vec<String>,
}

#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ContentDetails {
    #[serde(deserialize_with = "iso8601_duration")]
    pub duration: Duration,
    pub definition: Definition,
}

#[derive(Debug, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum Definition {
    Sd,
    Hd,
    #[serde(other)]
    Unknown,
}

#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct Statistics {
    #[serde(default, deserialize_with = "stringy_u64")]
    pub view_count: u64,
    #[serde(default, deserialize_with = "stringy_u64")]
    pub like_count: u64,
}

fn stringy_u64<'de, D: Deserializer<'de>>(d: D) -> Result<u64, D::Error> {
    let s = String::deserialize(d)?;
    s.parse().map_err(serde::de::Error::custom)
}

/// PT1H4M13S -> 3853s. Good enough: this API never emits days or fractions.
fn iso8601_duration<'de, D: Deserializer<'de>>(d: D) -> Result<Duration, D::Error> {
    let s = String::deserialize(d)?;
    let body = s
        .strip_prefix("PT")
        .ok_or_else(|| serde::de::Error::custom(format!("not a PT duration: {s}")))?;
    let (mut secs, mut acc) = (0u64, 0u64);
    for ch in body.chars() {
        match ch {
            '0'..='9' => acc = acc * 10 + (ch as u64 - '0' as u64),
            'H' => { secs += acc * 3600; acc = 0 }
            'M' => { secs += acc * 60; acc = 0 }
            'S' => { secs += acc; acc = 0 }
            _ => return Err(serde::de::Error::custom(format!("bad unit {ch} in {s}"))),
        }
    }
    Ok(Duration::from_secs(secs))
}

The #[serde(other)] variant on Definition is the one I would push hardest on. An upstream that adds a new enum value should not take your nightly cron down. Anywhere the value is descriptive rather than load-bearing, give it an Unknown arm and move on.

What I deliberately did not do is put deny_unknown_fields on these structs. In production, unknown fields are fine — the upstream adds them constantly. In tests, they are a signal. More on that later.

The client, and the fact that quota is state

We run three API keys. Each has a daily quota, and a trending list costs one unit while a search costs a hundred, so the interesting failure is not a network error — it is a 403 with quotaExceeded in the body at 04:00 UTC.

The naive design round-robins keys per request. Do not do that. Round-robin spreads your consumption evenly, which means all three keys hit their ceiling within the same few minutes and the run dies with a third of the work left. Sticky-until-burned is better: use key one until it returns quotaExceeded, retire it for the process lifetime, advance to key two, and retry the same request immediately with no backoff, because there is nothing to back off from.

A few reqwest specifics that matter more than they look:

reqwest::Client is the connection pool. Build it once, clone() it everywhere — clones share the pool. Building one per request costs you a fresh TLS handshake every time, which is the exact thing you left PHP to avoid.
Set connect_timeout separately from timeout. A hung DNS resolution and a slow 200 need different patience.
gzip(true) on a listing endpoint that returns descriptions and tags is a real bandwidth win on a shared host.
Use std::sync::Mutex, not tokio::sync::Mutex, for the key ring. The lock is never held across an .await — look at the scoping in get() below. A std mutex there is cheaper and the borrow checker keeps you honest.

use reqwest::{Client, StatusCode};
use std::sync::{Arc, Mutex};
use std::time::Duration;

pub struct KeyRing {
    keys: Vec<String>,
    cursor: usize,
}

impl KeyRing {
    pub fn len(&self) -> usize { self.keys.len() }

    pub fn current(&self) -> Option<String> {
        self.keys.get(self.cursor).cloned()
    }

    /// Retire the key we just used, but only if another task has not
    /// already advanced past it. Without this guard, three concurrent
    /// 403s burn three keys instead of one.
    pub fn retire(&mut self, used: &str) {
        if self.keys.get(self.cursor).map(String::as_str) == Some(used) {
            self.cursor += 1;
        }
    }
}

#[derive(Clone)]
pub struct VideoApi {
    http: Client,
    base: String,
    keys: Arc<Mutex<KeyRing>>,
}

impl VideoApi {
    pub fn new(keys: Vec<String>, base: impl Into<String>) -> anyhow::Result<Self> {
        let http = Client::builder()
            .user_agent(concat!("tvh-sdk/", env!("CARGO_PKG_VERSION")))
            .timeout(Duration::from_secs(15))
            .connect_timeout(Duration::from_secs(5))
            .pool_idle_timeout(Duration::from_secs(90))
            .pool_max_idle_per_host(8)
            .gzip(true)
            .build()?;
        Ok(Self {
            http,
            base: base.into(),
            keys: Arc::new(Mutex::new(KeyRing { keys, cursor: 0 })),
        })
    }

    pub async fn trending(
        &self,
        region: &str,
        page: Option<&str>,
    ) -> Result<Page<Video>, ApiError> {
        let mut params: Vec<(&str, String)> = vec![
            ("part", "snippet,contentDetails,statistics".into()),
            ("chart", "mostPopular".into()),
            ("regionCode", region.to_string()),
            ("maxResults", "50".into()),
        ];
        if let Some(t) = page {
            params.push(("pageToken", t.to_string()));
        }
        self.get("/videos", params).await
    }
}

The base field is not there for elegance. It is there so the integration tests can point the whole SDK at a local mock server without a feature flag or a trait object.

Errors that tell you what to change

This is the part that paid for the rewrite, and it has nothing to do with speed.

When serde fails to decode a response, the default error is invalid type: string "1048576", expected u64 at line 1 column 4171. The payload is one long line, so the column is useless on its own and you are left re-running the request by hand to see what happened. So the decode error carries a truncated copy of the raw body. Two kilobytes in a log line has caught every schema change the upstream has thrown at us since.

Truncating that body is where you meet the CJK tax: &s[..2048] panics if byte 2048 lands in the middle of a multi-byte character, and with Japanese and Korean titles it lands there roughly two times in three. Walk back to a char boundary.

#[derive(Debug, thiserror::Error)]
pub enum ApiError {
    #[error("all {0} API keys are out of quota")]
    QuotaExhausted(usize),
    #[error("upstream returned {status} for {path}: {body}")]
    Upstream { status: StatusCode, path: String, body: String },
    #[error("decode failed for {path}: {source} -- raw: {raw}")]
    Decode { path: String, #[source] source: serde_json::Error, raw: String },
    #[error(transparent)]
    Transport(#[from] reqwest::Error),
}

fn truncate(s: &str, max: usize) -> String {
    if s.len() <= max {
        return s.to_string();
    }
    let mut end = max;
    while !s.is_char_boundary(end) {
        end -= 1;
    }
    format!("{}...", &s[..end])
}

fn jitter() -> Duration {
    Duration::from_millis(fastrand::u64(0..120))
}

impl VideoApi {
    async fn get<T: serde::de::DeserializeOwned>(
        &self,
        path: &str,
        params: Vec<(&str, String)>,
    ) -> Result<T, ApiError> {
        let url = format!("{}{}", self.base, path);
        let mut attempt = 0u32;

        loop {
            // Lock scope ends here; nothing is held across the await below.
            let (key, total) = {
                let ring = self.keys.lock().unwrap();
                (ring.current(), ring.len())
            };
            let key = key.ok_or(ApiError::QuotaExhausted(total))?;

            let resp = self
                .http
                .get(&url)
                .query(&params)
                .query(&[("key", &key)])
                .send()
                .await?;

            let status = resp.status();
            if status.is_success() {
                let raw = resp.text().await?;
                return serde_json::from_str::<T>(&raw).map_err(|source| ApiError::Decode {
                    path: path.to_string(),
                    source,
                    raw: truncate(&raw, 2048),
                });
            }

            let body = resp.text().await.unwrap_or_default();

            if status == StatusCode::FORBIDDEN && body.contains("quotaExceeded") {
                self.keys.lock().unwrap().retire(&key);
                continue; // same request, next key, no backoff
            }

            let retryable = status == StatusCode::TOO_MANY_REQUESTS || status.is_server_error();
            if retryable && attempt < 4 {
                tokio::time::sleep(Duration::from_millis(250u64 << attempt) + jitter()).await;
                attempt += 1;
                continue;
            }

            return Err(ApiError::Upstream {
                status,
                path: path.to_string(),
                body: truncate(&body, 512),
            });
        }
    }
}

Note what is not retried: a 400, a 404, a 403 that is not a quota problem. Those are bugs in your request, and retrying them four times with backoff only means you find out about them eight seconds later.

Concurrency sized for the upstream, not the CPU

Nine regions, up to four pages each. The obvious move is join_all over the regions, which gives you nine concurrent requests hammering a single API key, and the upstream answers with 429 until you stop.

The right shape falls out of the data: pagination is inherently sequential within a region because you need nextPageToken from page N to ask for page N+1, and completely independent across regions. So: sequential inside, bounded-parallel outside. buffer_unordered(3) was the number where throughput stopped improving and 429s had not yet started.

The other job of this stage is normalization, and for an Asia-Pacific catalogue that means Unicode. Titles arrive with full-width Latin (ＡＫＢ４８), half-width katakana (ｱｲｳ), and a generous spread of variation selectors. If you push those straight into SQLite, your FTS5 index contains three spellings of the same token and users searching the normal way find nothing. Running NFKC once in the SDK — at the point where you already own the string — means the PHP search path never has to think about it.

use futures::stream::{self, StreamExt};
use std::io::{BufWriter, Write};
use unicode_normalization::UnicodeNormalization;

const REGIONS: [&str; 9] = ["US", "GB", "JP", "KR", "TW", "SG", "VN", "TH", "HK"];

#[derive(serde::Serialize)]
struct Row {
    id: String,
    region: &'static str,
    title: String,
    /// NFKC-folded copy that feeds the FTS5 index.
    title_norm: String,
    channel_id: String,
    duration_s: u64,
    views: u64,
    published_at: String,
}

impl Row {
    fn from_video(region: &'static str, v: Video) -> Self {
        Row {
            id: v.id,
            region,
            title_norm: v.snippet.title.nfkc().collect::<String>().to_lowercase(),
            title: v.snippet.title,
            channel_id: v.snippet.channel_id,
            duration_s: v.content_details.map(|c| c.duration.as_secs()).unwrap_or(0),
            views: v.statistics.map(|s| s.view_count).unwrap_or(0),
            published_at: v.snippet.published_at.to_rfc3339(),
        }
    }
}

#[tokio::main(flavor = "multi_thread", worker_threads = 2)]
async fn main() -> anyhow::Result<()> {
    let api = VideoApi::new(load_keys()?, "https://www.googleapis.com/youtube/v3")?;

    let results = stream::iter(REGIONS)
        .map(|region| {
            let api = api.clone();
            async move {
                let mut rows = Vec::new();
                let mut token: Option<String> = None;
                for _ in 0..4 {
                    let page = api.trending(region, token.as_deref()).await?;
                    rows.extend(page.items.into_iter().map(|v| Row::from_video(region, v)));
                    match page.next_page_token {
                        Some(t) => token = Some(t),
                        None => break,
                    }
                }
                Ok::<_, ApiError>(rows)
            }
        })
        .buffer_unordered(3)
        .collect::<Vec<_>>()
        .await;

    let stdout = std::io::stdout();
    let mut out = BufWriter::new(stdout.lock());
    let (mut emitted, mut failed) = (0usize, 0usize);

    for result in results {
        match result {
            Ok(rows) => {
                for row in rows {
                    serde_json::to_writer(&mut out, &row)?;
                    out.write_all(b"\n")?;
                    emitted += 1;
                }
            }
            Err(e) => {
                failed += 1;
                eprintln!("region failed: {e:#}");
            }
        }
    }
    out.flush()?;
    eprintln!("emitted={emitted} failed_regions={failed}");

    // A partial run is fine; an empty one is a failure the cron should see.
    if emitted == 0 {
        std::process::exit(3);
    }
    Ok(())
}

Two worker threads, not num_cpus. This workload is entirely IO-bound and it shares a box with PHP-FPM; there is no reason to spin up eight schedulers to wait on sockets.

The PHP side stays boring

NDJSON over a pipe, one row per line, ingested in batched transactions. No HTTP between the two, no daemon to supervise, no port to firewall. The Rust binary is just a program the cron shells out to.

<?php
declare(strict_types=1);

$cmd  = escapeshellcmd(__DIR__ . '/bin/tvh-fetch');
$spec = [1 => ['pipe', 'w'], 2 => ['pipe', 'w']];
$proc = proc_open($cmd, $spec, $pipes);
if (!is_resource($proc)) {
    throw new RuntimeException('fetch binary did not start');
}

$db = new PDO('sqlite:' . __DIR__ . '/../data/app.db');
$db->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
$db->exec('PRAGMA journal_mode=WAL');

$stmt = $db->prepare(
    'INSERT INTO videos (id, region, title, title_norm, channel_id, duration_s, views, published_at)
     VALUES (:id, :region, :title, :title_norm, :channel_id, :duration_s, :views, :published_at)
     ON CONFLICT(id, region) DO UPDATE SET
        views = excluded.views,
        title = excluded.title,
        title_norm = excluded.title_norm'
);

$db->beginTransaction();
$n = 0;
while (($line = fgets($pipes[1])) !== false) {
    $row = json_decode($line, true, 16, JSON_THROW_ON_ERROR);
    $stmt->execute([
        ':id' => $row['id'],
        ':region' => $row['region'],
        ':title' => $row['title'],
        ':title_norm' => $row['title_norm'],
        ':channel_id' => $row['channel_id'],
        ':duration_s' => $row['duration_s'],
        ':views' => $row['views'],
        ':published_at' => $row['published_at'],
    ]);
    if (++$n % 500 === 0) {
        $db->commit();
        $db->beginTransaction();
    }
}
$db->commit();

// Drain stdout fully and close it BEFORE reading stderr, or a chatty
// binary can fill the stderr pipe buffer and deadlock both processes.
fclose($pipes[1]);
$err  = stream_get_contents($pipes[2]);
fclose($pipes[2]);
$code = proc_close($proc);

if ($code !== 0) {
    throw new RuntimeException("fetch exited {$code}: {$err}");
}
fprintf(STDERR, "ingested %d rows\n", $n);

That pipe-ordering comment is not theoretical. The first version read stderr inside the loop and hung on a run where one region produced a few hundred kilobytes of retry warnings.

Testing an SDK you do not control the server for

Three layers, cheapest first:

Fixture decoding. Record real responses once into tests/fixtures/*.json and write a single test that walks the directory and runs serde_json::from_str::<Page<Video>> over each. It takes milliseconds and it is the test that fails when you refresh fixtures against a changed upstream.
Drift detection. In a test-only module, mirror the structs with #[serde(deny_unknown_fields)] and assert against the same fixtures. Production stays permissive; the test tells you when a new field appeared so you can decide whether you want it.
Behaviour against a mock. wiremock gives you a MockServer, and because base is injectable you pass its URI to VideoApi::new. The tests worth writing here are the failure paths: a 403 quotaExceeded on key one followed by a 200 on key two (assert both that the result is Ok and that exactly two requests were made), a 500 that succeeds on the third attempt, and a 400 that does not retry at all.

That last assertion — request count, not just the result — is the one that catches a retry loop you accidentally made infinite.

What it actually bought

The nine-region run went from about 38 minutes to 41 seconds. The binary is 4.8 MB built against musl for the shared host, and peaks around 12 MB RSS. Those numbers are nice and they are not the point.

The point is the three bugs the type system surfaced during the port, all of which had been quietly wrong in PHP for months: view counts compared as strings so sorting was lexicographic past a certain magnitude, live streams with no contentDetails producing a duration of zero that the UI rendered as 0:00, and a null defaultAudioLanguage mislabelling a chunk of the Taiwanese feed. None of those threw. They just produced slightly wrong pages.

The honest costs: a cross-compile step in the deploy, a binary in an otherwise text-only FTP push, and a second language in a codebase maintained by one person. If you pull one region and 200 videos a day, this is not worth it. Our threshold was concrete — the cron stopped finishing inside its own interval.

Conclusion

The reqwest and serde part of this is maybe 200 lines and it is the least interesting thing in the repo. What matters is where you put the boundary: decode into real types at the edge, normalize Unicode once while you still own the string, keep quota as explicit state instead of a retry loop's side effect, and attach the raw payload to decode errors so a schema change is a five-minute fix instead of an afternoon. Do that, and the rest of the stack — PHP, SQLite, whatever it is — gets to stay boring, which is the actual goal.