Mirror-bot/teapawt
1
1 Fork 0
mirror of https://github.com/4cuck/teapawt.git synced 2026-10-28 05:05:04 +00:00
Code Issues Projects Releases Activity

Park proxy exits X slow-walks, copy calls that run long, and fetch a profile tab's user and posts together.

Browse Source 
An exit whose share of slow answers runs far above the rest of the fleet is
parked for two minutes, doubling while it keeps relapsing, and its accounts
use a healthy exit meanwhile. A call still unanswered at the larger of its
operation's 95th percentile and twice its median gets one copy on another
account and exit, and the first answer wins.

Profile tabs, the posts menu, user search, multi-user timelines and user RSS
send UserByScreenName beside the timeline when the id is stored. A tab gives
a slow photo strip 60 ms after the timeline before rendering without it.

Pages are gzipped before they cross the tunnel to Cloudflare. Logs go through
a non-blocking writer, so a lagging pm2 drops lines instead of stalling
requests, and stall warnings leave the feature flags out of the URL. Answer
times per operation and the parked exits are logged every minute.

Co-authored-by: Cursor <[email protected]>
This commit is contained in:
4cuckandCursor committed 2026-10-27 01:52:54 +00:00
1 parent 7438d9c867
commit 0988cae123
12 files changed
+1160 -192
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

Cargo.lock
Generated
+35
View File
@@ -394,6 +394,21 @@ dependencies = [
"cfg-if",
]
[[package]]
name = "crossbeam-channel"
version = "0.5.17"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "98b0cc327b5bc766e7fda9c9260cc0fa81b43a8e240440422dff70788e3f9ef1"
dependencies = [
"crossbeam-utils",
]
[[package]]
name = "crossbeam-utils"
version = "0.8.23"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a31eee39dddec8330830986fcd7625edb5a24ec90ea038215273bbc3adb08ac6"
[[package]]
name = "crypto-common"
version = "0.2.2"
@@ -1756,6 +1771,12 @@ version = "2.6.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292"
[[package]]
name = "symlink"
version = "0.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a7973cce6668464ea31f176d85b13c7ab3bba2cb3b77a2ed26abd7801688010a"
[[package]]
name = "syn"
version = "2.0.119"
@@ -1829,6 +1850,7 @@ dependencies = [
"tower",
"tower-http 0.7.0",
"tracing",
"tracing-appender",
"tracing-subscriber",
"xitter-txid",
]
@@ -2132,6 +2154,19 @@ dependencies = [
"tracing-core",
]
[[package]]
name = "tracing-appender"
version = "0.2.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "050686193eb999b4bb3bc2acfa891a13da00f79734704c4b8b4ef1a10b368a3c"
dependencies = [
"crossbeam-channel",
"symlink",
"thiserror",
"time",
"tracing-subscriber",
]
[[package]]
name = "tracing-attributes"
version = "0.1.31"
Cargo.toml
+1
View File
@@ -71,6 +71,7 @@ tower-http = {
version = "0.7.0",
}
tracing = "0.1.44"
tracing-appender = "0.2.5"
tracing-subscriber = { features = [ "env-filter" ], version = "0.3.23" }
xitter-txid = "0.2.0"
src/api/auth.rs
+96 -54
View File
@@ -4,6 +4,7 @@ use std::{
fmt::Write as _,
ops::Deref,
path::Path,
result::Result as StdResult,
sync::{
Arc,
atomic::{AtomicBool, AtomicUsize, Ordering},
@@ -93,6 +94,16 @@ struct LimitState {
const STATE_FLUSH_INTERVAL: Duration = Duration::from_secs(15);
/// Why one pass over the pool found no account.
enum Miss {
/// None can serve this call: rejected, refused, or not acceptable.
Nobody,
/// Every candidate is out of quota.
Spent,
/// A candidate has quota but is at its concurrency cap.
Busy,
}
struct SessionSlot {
credentials: Arc<SessionCredentials>,
permits: Arc<Semaphore>,
@@ -404,61 +415,14 @@ impl SessionPool {
}
loop {
let start = self.cursor.fetch_add(1, Ordering::Relaxed);
let mut limits = self.limits.write().await;
let count = self.sessions.len();
// Verified accounts go first, still in rotation. Stop at the first
// one with a free slot instead of listing every account first: this
// lock is taken on every call to X, and walking the whole pool made
// later requests wait out the scan.
let mut busy = false;
let mut any = false;
for verified_only in [true, false] {
for offset in 0..count {
let index = start.wrapping_add(offset) % count;
let slot = &self.sessions[index];
if excluded_id == Some(slot.credentials.id)
|| required_kind.is_some_and(|kind| slot.credentials.kind != kind)
{
continue;
}
let api = api_for(slot.credentials.kind);
let (rejected, verified) = match limits.get(&slot.credentials.id) {
Some(session_limits) => {
(session_limits.rejected, session_limits.verified(api))
},
None => (false, false),
};
if rejected || verified != verified_only {
continue;
}
any = true;
let spent = match limits.get_mut(&slot.credentials.id) {
Some(session_limits) => session_limits.try_spend(api),
None => Spend::Unmetered,
};
if spent == Spend::Denied {
continue;
}
let metered = spent == Spend::Metered;
if let Ok(permit) = Arc::clone(&slot.permits).try_acquire_owned() {
let credentials = Arc::clone(&slot.credentials);
return Ok(self.lease(credentials, permit, metered));
}
if metered
&& let Some(session_limits) = limits.get_mut(&slot.credentials.id)
{
session_limits.refund_spend(api);
}
busy = true;
}
}
if !any {
return Err(Error::NoSessions);
}
if !busy {
return Err(Error::RateLimited);
match self.scan(&mut limits, &api_for, required_kind, &|id| {
excluded_id != Some(id)
}) {
Ok(lease) => return Ok(lease),
Err(Miss::Nobody) => return Err(Error::NoSessions),
Err(Miss::Spent) => return Err(Error::RateLimited),
Err(Miss::Busy) => {},
}
let waiter = self.freed.notified();
tokio::pin!(waiter);
@@ -470,6 +434,84 @@ impl SessionPool {
}
}
/// One pass for a spare account that `accept` allows, taken only if it
/// has a free slot now. Used for the second copy of a slow call, which is
/// not worth waiting for.
pub(crate) async fn try_acquire_where(
&self,
api: &str,
required_kind: Option<SessionKind>,
accept: impl Fn(i64) -> bool + Send,
) -> Option<SessionLease> {
if self.sessions.is_empty() {
return None;
}
let mut limits = self.limits.write().await;
self
.scan(&mut limits, &|_| api, required_kind, &accept)
.ok()
}
/// Verified accounts go first, still in rotation. Stop at the first one
/// with a free slot instead of listing every account first: this lock is
/// taken on every call to X, and walking the whole pool made later
/// requests wait out the scan.
fn scan<'a>(
&self,
limits: &mut HashMap<i64, SessionLimits>,
api_for: &impl Fn(SessionKind) -> &'a str,
required_kind: Option<SessionKind>,
accept: &impl Fn(i64) -> bool,
) -> StdResult<SessionLease, Miss> {
let start = self.cursor.fetch_add(1, Ordering::Relaxed);
let count = self.sessions.len();
let mut busy = false;
let mut any = false;
for verified_only in [true, false] {
for offset in 0..count {
let index = start.wrapping_add(offset) % count;
let slot = &self.sessions[index];
if !accept(slot.credentials.id)
|| required_kind.is_some_and(|kind| slot.credentials.kind != kind)
{
continue;
}
let api = api_for(slot.credentials.kind);
let (rejected, verified) = match limits.get(&slot.credentials.id) {
Some(session_limits) => (session_limits.rejected, session_limits.verified(api)),
None => (false, false),
};
if rejected || verified != verified_only {
continue;
}
any = true;
let spent = match limits.get_mut(&slot.credentials.id) {
Some(session_limits) => session_limits.try_spend(api),
None => Spend::Unmetered,
};
if spent == Spend::Denied {
continue;
}
let metered = spent == Spend::Metered;
if let Ok(permit) = Arc::clone(&slot.permits).try_acquire_owned() {
let credentials = Arc::clone(&slot.credentials);
return Ok(self.lease(credentials, permit, metered));
}
if metered && let Some(session_limits) = limits.get_mut(&slot.credentials.id) {
session_limits.refund_spend(api);
}
busy = true;
}
}
Err(if !any {
Miss::Nobody
} else if busy {
Miss::Busy
} else {
Miss::Spent
})
}
fn lease(
&self,
credentials: Arc<SessionCredentials>,
src/api/client.rs
+159 -4
View File
@@ -1,14 +1,21 @@
use std::sync::Arc;
use std::{
sync::Arc,
time::{Duration, Instant},
};
use axum::http::header;
use serde::{Deserialize, de::DeserializeOwned};
use tokio::time::timeout;
use tokio::{
sync::Semaphore,
time::{interval, sleep, timeout},
};
use super::{
ProxyPool, SessionLease, SessionPool, TidClient,
budget::{self, ClientBudget},
endpoints,
http::{Egress, HttpClient, ProxyConfig, Purpose},
latency::Latency,
operations::Operations,
parser,
};
@@ -56,6 +63,28 @@ const GRAPHQL_REPLACEMENTS: usize = 2;
/// Per-account 429 (JSON / `x-rate-limit-*`). HTML 429s are the WAF and must
/// not walk the pool — that used to zero every cookie for 15 minutes.
const RATE_LIMIT_REPLACEMENTS: usize = 5;
/// Second copies of slow calls in flight at once. In a steady state about
/// one call in ten is hedged; this cap keeps a sudden slowdown at X from
/// doubling every call while the latency window catches up.
const MAX_HEDGES: usize = 48;
/// How often the upstream latency summary is logged.
const LATENCY_REPORT_EVERY: Duration = Duration::from_secs(60);
/// An answer about the thing asked for, which a second account would give
/// too. Anything else is the account or the path, and the other copy of a
/// hedged call is still worth waiting for.
const fn settles_the_call(err: &Error) -> bool {
matches!(
err,
Error::NotFound(_)
| Error::UserNotFound(_)
| Error::UserSuspended(_)
| Error::ProtectedUser(_)
| Error::TweetNotFound(_)
| Error::InvalidUrl(_)
| Error::SearchDenylisted
)
}
/// Edge 502/503/504, including an empty body and a stream the proxy reset.
fn is_gateway_blip(message: &str) -> bool {
@@ -290,6 +319,8 @@ pub struct ApiClient {
budget: ClientBudget,
tid_enabled: bool,
denylist: super::search_denylist::SearchDenylist,
latency: Arc<Latency>,
hedges: Arc<Semaphore>,
}
impl ApiClient {
@@ -338,9 +369,39 @@ impl ApiClient {
budget: ClientBudget::new(config.config.client_budget),
tid_enabled: !config.config.disable_tid,
denylist: super::search_denylist::SearchDenylist::load(denylist_path),
latency: Arc::new(Latency::default()),
hedges: Arc::new(Semaphore::new(MAX_HEDGES)),
}
}
/// Log how long X takes per operation, how often a second copy went out
/// and won, and which exits are parked.
pub fn spawn_latency_report(&self) {
let api = self.clone();
tokio::spawn(async move {
let mut every = interval(LATENCY_REPORT_EVERY);
every.tick().await;
loop {
every.tick().await;
for line in api.latency.report() {
tracing::info!("upstream {line}");
}
let parked = api.client.parked_exits();
if !parked.is_empty() {
tracing::info!(?parked, "proxy exits parked (port, seconds left)");
}
}
});
}
/// The proxy port this account is pinned to, when there is a SOCKS pool.
fn pinned_port(&self, session_id: i64) -> Option<u16> {
self
.proxies
.as_ref()
.map(|pool| pool.port_for(session_id))
}
/// The browser and proxy a request for this session goes out as.
pub(crate) fn egress_for(&self, session: &SessionLease) -> Egress {
Egress {
@@ -479,9 +540,10 @@ impl ApiClient {
where
T: DeserializeOwned,
{
let mut last = self
.graphql_request_inner(&session, endpoint, variables, features, field_toggles)
let (answered_on, mut last) = self
.hedged_attempt(session, endpoint, variables, features, field_toggles, retry_kind)
.await;
session = answered_on;
// Quota failures may walk RATE_LIMIT_REPLACEMENTS accounts. The other
// cases stay capped at GRAPHQL_REPLACEMENTS inside the match.
@@ -577,6 +639,99 @@ impl ApiClient {
last
}
/// The first attempt of a call, plus a second copy on another account and
/// exit once it has run past most of its operation's recent answers.
/// The first good answer is kept and the other copy is dropped. The copy
/// is not billed to the caller: it buys latency, not another page.
async fn hedged_attempt<T>(
&self,
session: SessionLease,
endpoint: &str,
variables: &str,
features: &str,
field_toggles: Option<&str>,
retry_kind: Option<SessionKind>,
) -> (SessionLease, Result<T>)
where
T: DeserializeOwned,
{
let first_id = session.id;
let delay = self.latency.hedge_delay(endpoint);
// Boxed: two request futures side by side would double the size of
// every handler future that awaits this one.
let mut first =
Box::pin(self.timed_attempt(session, endpoint, variables, features, field_toggles));
tokio::select! {
biased;
done = &mut first => return done,
() = sleep(delay) => {},
}
let Ok(permit) = Arc::clone(&self.hedges).try_acquire_owned() else {
return first.await;
};
let first_port = self.pinned_port(first_id);
let Some(spare) = self
.sessions
.try_acquire_where(endpoint, retry_kind, |id| {
id != first_id && (first_port.is_none() || self.pinned_port(id) != first_port)
})
.await
else {
return first.await;
};
self.latency.hedged(endpoint);
let mut second =
Box::pin(self.timed_attempt(spare, endpoint, variables, features, field_toggles));
let outcome = tokio::select! {
done = &mut first => {
if done.1.as_ref().is_err_and(|err| !settles_the_call(err)) {
let other = second.await;
if other.1.is_ok() {
self.latency.rescued(endpoint);
other
} else {
done
}
} else {
done
}
},
done = &mut second => {
if done.1.is_ok() {
self.latency.rescued(endpoint);
done
} else {
first.await
}
},
};
drop(permit);
outcome
}
/// One attempt that hands its lease back, timing the ones X answers.
async fn timed_attempt<T>(
&self,
session: SessionLease,
endpoint: &str,
variables: &str,
features: &str,
field_toggles: Option<&str>,
) -> (SessionLease, Result<T>)
where
T: DeserializeOwned,
{
let started = Instant::now();
let result = self
.graphql_request_inner(&session, endpoint, variables, features, field_toggles)
.await;
if result.is_ok() {
self.latency.record(endpoint, started.elapsed());
}
(session, result)
}
/// Take a session for one upstream call, billing the caller first.
///
/// The only way to reach X, so a caller cannot spend quota by taking a
src/api/http.rs
+355 -15
View File
@@ -18,7 +18,7 @@ use std::{
atomic::{AtomicU64, AtomicUsize, Ordering},
},
task::{Context, Poll},
time::{Duration, SystemTime, UNIX_EPOCH},
time::{Duration, Instant, SystemTime, UNIX_EPOCH},
};
use axum::http::{HeaderMap, HeaderName, HeaderValue, Method, StatusCode, header};
@@ -63,6 +63,27 @@ const DIRECT_BLOCK_FOR: Duration = Duration::from_secs(5 * 60);
/// An X/Cloudflare HTML 429 is tied to the public IP, not the account. Park
/// that SOCKS exit long enough for the edge window to clear.
const EXIT_BLOCK_FOR: Duration = Duration::from_secs(5 * 60);
/// Headers later than this count against the exit that carried the call.
const SLOW_HEADERS: Duration = Duration::from_secs(1);
/// A call dropped unanswered this late, because another copy answered
/// first, counts against its exit too.
const ABANDONED_SLOW: Duration = Duration::from_millis(700);
/// An exit's answers are counted in two buckets of this length.
const HEALTH_BUCKET_MS: u64 = 30_000;
/// Slow answers an exit needs in its window before it can be parked.
const SLOW_EXIT_MIN: u32 = 8;
/// Share of the exit's answers that have to be slow. A healthy exit runs a
/// few percent; one X slow-walks ran about half for a whole day.
const SLOW_EXIT_SHARE: f64 = 0.25;
/// And that share against the fleet's median. When X is slow for every exit,
/// none of them is the problem.
const SLOW_EXIT_VS_FLEET: f64 = 3.0;
/// First parking for slowness. It doubles while the exit keeps relapsing.
const SLOW_PARK_FIRST: Duration = Duration::from_mins(2);
const SLOW_PARK_MAX: Duration = Duration::from_mins(30);
/// An exit that stays healthy this long after its parking ends starts over
/// at [`SLOW_PARK_FIRST`].
const SLOW_PARK_MEMORY_MS: u64 = 10 * 60 * 1000;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ProxyKind {
@@ -154,16 +175,158 @@ impl FallbackPool {
}
}
/// One exit's recent answers, in two half-minute buckets so the count always
/// covers the last thirty to sixty seconds.
#[derive(Clone, Copy, Default)]
struct ExitSamples {
bucket: u64,
fast: [u32; 2],
slow: [u32; 2],
}
impl ExitSamples {
const fn roll(&mut self, now: u64) {
let bucket = now / HEALTH_BUCKET_MS;
match bucket.saturating_sub(self.bucket) {
0 => {},
1 => {
self.fast = [0, self.fast[0]];
self.slow = [0, self.slow[0]];
},
_ => {
self.fast = [0, 0];
self.slow = [0, 0];
},
}
self.bucket = bucket;
}
/// Fast and slow answers in the window.
const fn totals(&self) -> (u32, u32) {
(
self.fast[0].saturating_add(self.fast[1]),
self.slow[0].saturating_add(self.slow[1]),
)
}
}
/// The stable SOCKS exits used by cookie sessions. A session normally stays
/// pinned, but temporarily moves to another healthy exit when X rate-limits
/// its public IP.
/// its public IP or answers it far slower than the other exits.
struct PrimaryPool {
proxies: Vec<ProxyConfig>,
tags: HashSet<u64>,
blocked_until: Mutex<HashMap<u64, u64>>,
samples: Mutex<HashMap<u64, ExitSamples>>,
/// Slow parkings in a row per exit, and when the last one ends.
relapses: Mutex<HashMap<u64, (u32, u64)>>,
}
impl PrimaryPool {
fn new(proxies: Vec<ProxyConfig>, tags: HashSet<u64>) -> Self {
Self {
proxies,
tags,
blocked_until: Mutex::new(HashMap::new()),
samples: Mutex::new(HashMap::new()),
relapses: Mutex::new(HashMap::new()),
}
}
/// Count one call against `proxy`: answered, or slow (late headers, no
/// headers, no connection). Returns the parking time when this call made
/// the exit one to avoid.
fn record(&self, proxy: &ProxyConfig, slow: bool) -> Option<Duration> {
let tag = proxy_tag(Some(proxy));
if !self.tags.contains(&tag) {
return None;
}
let now = unix_millis();
let mut samples = self.samples.lock().ok()?;
let mine = samples.entry(tag).or_default();
mine.roll(now);
if slow {
mine.slow[0] = mine.slow[0].saturating_add(1);
} else {
mine.fast[0] = mine.fast[0].saturating_add(1);
return None;
}
let (fast, slow_count) = mine.totals();
let share = f64::from(slow_count) / f64::from(fast.saturating_add(slow_count));
if slow_count < SLOW_EXIT_MIN || share < SLOW_EXIT_SHARE {
return None;
}
let mut shares = samples
.values_mut()
.filter_map(|other| {
other.roll(now);
let (quick, late) = other.totals();
let total = quick.saturating_add(late);
(total >= SLOW_EXIT_MIN).then(|| f64::from(late) / f64::from(total))
})
.collect::<Vec<_>>();
shares.sort_by(f64::total_cmp);
let median = shares.get(shares.len() / 2).copied().unwrap_or(0.0);
if share < SLOW_EXIT_VS_FLEET * median {
return None;
}
samples.insert(tag, ExitSamples::default());
drop(samples);
self.park_slow(tag, now)
}
fn park_slow(&self, tag: u64, now: u64) -> Option<Duration> {
let mut blocked = self.blocked_until.lock().ok()?;
blocked.retain(|_, until| *until > now);
// Parked already, or so many are parked that the rest would carry the
// whole pool.
if blocked.contains_key(&tag) || blocked.len() >= (self.proxies.len() / 3).max(1) {
return None;
}
let (park, until) = self.next_parking(tag, now)?;
blocked.insert(tag, until);
drop(blocked);
Some(park)
}
/// How long to park this exit, doubled when it relapsed soon after its
/// last parking ended, and the moment that ends.
fn next_parking(&self, tag: u64, now: u64) -> Option<(Duration, u64)> {
let mut relapses = self.relapses.lock().ok()?;
let (count, last_until) = relapses.get(&tag).copied().unwrap_or((0, 0));
let count = if last_until.saturating_add(SLOW_PARK_MEMORY_MS) > now {
count.saturating_add(1)
} else {
0
};
let park = SLOW_PARK_FIRST
.saturating_mul(1_u32 << count.min(8))
.min(SLOW_PARK_MAX);
let until = now.saturating_add(u64::try_from(park.as_millis()).unwrap_or(u64::MAX));
relapses.insert(tag, (count, until));
drop(relapses);
Some((park, until))
}
/// Parked exits by port, with the seconds each has left.
fn parked(&self) -> Vec<(u16, u64)> {
let now = unix_millis();
let Ok(blocked) = self.blocked_until.lock() else {
return Vec::new();
};
let mut parked = self
.proxies
.iter()
.filter_map(|proxy| {
let until = *blocked.get(&proxy_tag(Some(proxy)))?;
(until > now).then(|| (proxy.port, (until - now) / 1000))
})
.collect::<Vec<_>>();
drop(blocked);
parked.sort_unstable();
parked
}
fn contains(&self, proxy: &ProxyConfig) -> bool {
self.tags.contains(&proxy_tag(Some(proxy)))
}
@@ -334,15 +497,35 @@ impl HttpClient {
cooldown_secs = EXIT_BLOCK_FOR.as_secs(),
"proxy exit circuit breaker ready"
);
self.primary_pool = Some(Arc::new(PrimaryPool {
proxies,
tags,
blocked_until: Mutex::new(HashMap::new()),
}));
self.primary_pool = Some(Arc::new(PrimaryPool::new(proxies, tags)));
}
self
}
/// Primary exits parked right now, by port, with the seconds each has left.
pub fn parked_exits(&self) -> Vec<(u16, u64)> {
self
.primary_pool
.as_ref()
.map_or_else(Vec::new, |pool| pool.parked())
}
fn note_exit(&self, via: Option<&Egress>, slow: bool) {
let Some(pool) = self.primary_pool.as_ref() else {
return;
};
let Some(proxy) = via.and_then(|egress| egress.proxy.as_ref()) else {
return;
};
if let Some(park) = pool.record(proxy, slow) {
tracing::warn!(
port = proxy.port,
park_secs = park.as_secs(),
"X answers this proxy exit far slower than the rest; parking it"
);
}
}
fn has_fallback(&self) -> bool {
self.fallback_proxy.is_some() || self.fallback_pool.is_some()
}
@@ -719,11 +902,18 @@ impl HttpClient {
let inner = match tokio::time::timeout(MEDIA_HEADER_DEADLINE, send()).await {
Ok(Ok(inner)) => inner,
Ok(Err(err)) if is_transport_blip(&err) || err.is_connect() => {
tracing::warn!(uri, error = %err, "media fetch failed, retrying");
tracing::warn!(
uri = loggable(uri),
error = %describe(err.without_url()),
"media fetch failed, retrying"
);
send().await.map_err(describe)?
},
Err(_elapsed) => {
tracing::warn!(uri, "media headers stalled, opening a new connection");
tracing::warn!(
uri = loggable(uri),
"media headers stalled, opening a new connection"
);
send().await.map_err(describe)?
},
Ok(Err(err)) => return Err(describe(err)),
@@ -732,14 +922,27 @@ impl HttpClient {
}
if self.purpose == Purpose::Api && is_idempotent(method) {
let probe = ExitProbe::start(self, via);
let inner = match tokio::time::timeout(API_HEADER_DEADLINE, send()).await {
Ok(Ok(inner)) => inner,
Ok(Ok(inner)) => {
probe.answered();
inner
},
Ok(Err(err)) if is_transport_blip(&err) || err.is_connect() => {
tracing::warn!(uri, error = %err, "API connection failed, opening another");
probe.failed();
tracing::warn!(
uri = loggable(uri),
error = %describe(err.without_url()),
"API connection failed, opening another"
);
self.fresh_headers(method, uri, extra_headers, body, via).await?
},
Err(_elapsed) => {
tracing::warn!(uri, "API headers stalled, opening another connection");
probe.failed();
tracing::warn!(
uri = loggable(uri),
"API headers stalled, opening another connection"
);
match tokio::time::timeout(
API_HEADER_DEADLINE,
self.fresh_headers(method, uri, extra_headers, body, via),
@@ -750,12 +953,15 @@ impl HttpClient {
Ok(Err(err)) => return Err(err),
Err(_elapsed) => {
self.retire(via);
tracing::warn!(uri, "API headers stalled again");
tracing::warn!(uri = loggable(uri), "API headers stalled again");
return Err(Error::Internal("API headers stalled".into()));
},
}
},
Ok(Err(err)) => return Err(describe(err)),
Ok(Err(err)) => {
probe.failed();
return Err(describe(err));
},
};
return Ok(Response { inner });
}
@@ -792,7 +998,60 @@ impl HttpClient {
if !body.is_empty() {
request = request.body(body.clone());
}
request.send().await.map_err(describe)
let probe = ExitProbe::start(self, via);
match request.send().await {
Ok(inner) => {
probe.answered();
Ok(inner)
},
Err(err) => {
probe.failed();
Err(describe(err))
},
}
}
}
/// Counts one API call against the exit that carried it: when headers
/// arrive, when it fails, or when it is dropped unanswered because another
/// copy of the call answered first.
struct ExitProbe<'a> {
client: &'a HttpClient,
via: Option<&'a Egress>,
started: Instant,
settled: bool,
}
impl<'a> ExitProbe<'a> {
fn start(client: &'a HttpClient, via: Option<&'a Egress>) -> Self {
Self {
client,
via,
started: Instant::now(),
settled: false,
}
}
fn answered(mut self) {
let slow = self.started.elapsed() >= SLOW_HEADERS;
self.settle(slow);
}
fn failed(mut self) {
self.settle(true);
}
fn settle(&mut self, slow: bool) {
self.settled = true;
self.client.note_exit(self.via, slow);
}
}
impl Drop for ExitProbe<'_> {
fn drop(&mut self) {
if !self.settled && self.started.elapsed() >= ABANDONED_SLOW {
self.settle(true);
}
}
}
@@ -1054,6 +1313,13 @@ fn unix_millis() -> u64 {
.unwrap_or(0)
}
/// A GraphQL URI as logged: the operation and its variables. The feature and
/// field-toggle blobs after them are the same on every call and most of the
/// length, and every line of them is another write pm2 has to keep up with.
fn loggable(uri: &str) -> &str {
uri.find("&features=").map_or(uri, |at| &uri[..at])
}
/// A failure to send at all, as opposed to a timeout, a connect failure or a
/// bad response: the far end dropped a connection we thought was open.
fn is_transport_blip(err: &primp::Error) -> bool {
@@ -1244,6 +1510,80 @@ mod tests {
);
}
#[test]
fn a_logged_uri_keeps_its_variables_and_drops_the_feature_blob() {
let uri = "https://x.com/i/api/graphql/abc/TweetDetail?variables=%7B%22focalTweetId%22%3A%221%22%7D&features=%7B%22a%22%3Atrue%7D&fieldToggles=%7B%7D";
assert_eq!(
loggable(uri),
"https://x.com/i/api/graphql/abc/TweetDetail?variables=%7B%22focalTweetId%22%3A%221%22%7D"
);
let media = "https://pbs.twimg.com/media/abc.jpg?name=small";
assert_eq!(loggable(media), media);
}
#[test]
fn an_exit_x_answers_slowly_is_parked_and_the_rest_are_not() {
let proxies = (10001..=10006).map(proxy_on).collect::<Vec<_>>();
let client = HttpClient::new("", "", Purpose::Api).with_primary_proxies(proxies.clone());
let pool = client.primary_pool.as_ref().unwrap();
for proxy in &proxies {
for _ in 0..40 {
assert!(pool.record(proxy, false).is_none());
}
for _ in 0..2 {
assert!(pool.record(proxy, true).is_none());
}
}
let slow = &proxies[1];
let parked = (0..30).find_map(|_| pool.record(slow, true));
assert_eq!(parked, Some(SLOW_PARK_FIRST));
assert_eq!(client.parked_exits().len(), 1);
let pinned = Egress {
session: 7,
proxy: Some(slow.clone()),
};
assert!(client.primary_is_blocked(&pinned));
let moved = client.alternate_primary(&pinned).unwrap();
assert_ne!(moved.proxy.unwrap().port, slow.port);
let healthy = Egress {
session: 7,
proxy: Some(proxies[0].clone()),
};
assert!(!client.primary_is_blocked(&healthy));
}
#[test]
fn every_exit_slow_at_once_parks_none() {
let proxies = (10001..=10006).map(proxy_on).collect::<Vec<_>>();
let client = HttpClient::new("", "", Purpose::Api).with_primary_proxies(proxies.clone());
let pool = client.primary_pool.as_ref().unwrap();
for _ in 0..20 {
for proxy in &proxies {
assert!(pool.record(proxy, false).is_none());
assert!(pool.record(proxy, true).is_none());
}
}
assert!(client.parked_exits().is_empty());
}
#[test]
fn an_exit_that_relapses_is_parked_longer() {
let proxies = (10001..=10003).map(proxy_on).collect::<Vec<_>>();
let pool = PrimaryPool::new(
proxies.clone(),
proxies.iter().map(|proxy| proxy_tag(Some(proxy))).collect(),
);
let tag = proxy_tag(Some(&proxies[0]));
let now = unix_millis();
assert_eq!(pool.park_slow(tag, now), Some(SLOW_PARK_FIRST));
assert_eq!(pool.park_slow(tag, now), None, "already parked");
let after = now + SLOW_PARK_FIRST.as_millis() as u64 + 1;
assert_eq!(pool.park_slow(tag, after), Some(SLOW_PARK_FIRST * 2));
let healed = after + (SLOW_PARK_FIRST * 2).as_millis() as u64 + SLOW_PARK_MEMORY_MS + 1;
assert_eq!(pool.park_slow(tag, healed), Some(SLOW_PARK_FIRST));
}
#[test]
fn pool_exit_clients_are_shared_per_browser() {
let client = HttpClient::new("", "", Purpose::Api)
src/api/latency.rs
+224
View File
@@ -0,0 +1,224 @@
//! How long X takes to answer each operation, and when a call that has not
//! answered yet is worth a second copy on another account.
//!
//! A copy goes out once the first has run past most of its operation's
//! recent answers, so about one call in ten carries one. A path X answers
//! slowly then costs the reader that much more, not the 1.5 s header deadline
//! plus a second wait on a fresh connection to the same exit.
use std::{cmp::Reverse, collections::HashMap, fmt::Write as _, sync::Mutex, time::Duration};
/// Recent answers kept per operation. Enough for a steady 90th percentile,
/// few enough that the window follows X when it slows down.
const WINDOW: usize = 256;
/// Below this many answers the default delay is used.
const MIN_SAMPLES: usize = 32;
/// The delay is recomputed after this many new answers.
const RECOMPUTE_EVERY: u32 = 16;
/// Percent of recent answers that come back before a second copy is sent.
const HEDGE_PERCENTILE: usize = 95;
/// And never before this many medians. A call that is merely on the slow
/// side of normal is slow because of what was asked for, and a second
/// account takes just as long: at the 90th percentile, under 2% of
/// `TweetDetail` and `UserTweets` copies answered first.
const HEDGE_MEDIANS: u32 = 2;
/// Never sooner than this, so a fast operation is not copied on jitter.
const HEDGE_FLOOR: Duration = Duration::from_millis(250);
/// Out before the HTTP layer abandons the first connection at 1.5 s.
const HEDGE_CAP: Duration = Duration::from_millis(1200);
/// An operation with no history yet.
const HEDGE_DEFAULT: Duration = Duration::from_millis(800);
#[derive(Default)]
struct Operation {
/// Milliseconds per answered call, oldest overwritten first.
ring: Vec<u32>,
next: usize,
delay: Option<Duration>,
since_recompute: u32,
/// Since the last report.
answered: u64,
hedged: u64,
rescued: u64,
}
impl Operation {
fn push(&mut self, millis: u32) {
if self.ring.len() < WINDOW {
self.ring.push(millis);
} else {
self.ring[self.next] = millis;
}
self.next = (self.next + 1) % WINDOW;
self.answered += 1;
self.since_recompute += 1;
if self.ring.len() >= MIN_SAMPLES
&& (self.delay.is_none() || self.since_recompute >= RECOMPUTE_EVERY)
{
self.since_recompute = 0;
let slow = percentile(&self.ring, HEDGE_PERCENTILE)
.max(percentile(&self.ring, 50).saturating_mul(HEDGE_MEDIANS));
self.delay =
Some(Duration::from_millis(u64::from(slow)).clamp(HEDGE_FLOOR, HEDGE_CAP));
}
}
}
/// The value `percent` of `samples` fall at or below.
fn percentile(samples: &[u32], percent: usize) -> u32 {
if samples.is_empty() {
return 0;
}
let mut sorted = samples.to_vec();
let rank = (sorted.len() - 1) * percent.min(100) / 100;
let (_, value, _) = sorted.select_nth_unstable(rank);
*value
}
/// Operation name without the query id, so a new id X deploys keeps the
/// same history.
fn operation_name(endpoint: &str) -> &str {
endpoint.rsplit_once('/').map_or(endpoint, |(_, name)| name)
}
#[derive(Default)]
pub struct Latency {
operations: Mutex<HashMap<String, Operation>>,
}
impl Latency {
/// A call to `endpoint` came back with data after `took`.
pub fn record(&self, endpoint: &str, took: Duration) {
let millis = u32::try_from(took.as_millis()).unwrap_or(u32::MAX);
self.with(endpoint, |operation| operation.push(millis));
}
/// How long a call to `endpoint` may go unanswered before a second copy.
pub fn hedge_delay(&self, endpoint: &str) -> Duration {
let Ok(operations) = self.operations.lock() else {
return HEDGE_DEFAULT;
};
operations
.get(operation_name(endpoint))
.and_then(|operation| operation.delay)
.unwrap_or(HEDGE_DEFAULT)
}
/// A second copy went out.
pub fn hedged(&self, endpoint: &str) {
self.with(endpoint, |operation| operation.hedged += 1);
}
/// The second copy answered first.
pub fn rescued(&self, endpoint: &str) {
self.with(endpoint, |operation| operation.rescued += 1);
}
fn with(&self, endpoint: &str, update: impl FnOnce(&mut Operation)) {
let Ok(mut operations) = self.operations.lock() else {
return;
};
let name = operation_name(endpoint);
if let Some(operation) = operations.get_mut(name) {
update(operation);
} else {
let mut operation = Operation::default();
update(&mut operation);
operations.insert(name.to_owned(), operation);
}
}
/// One line per operation answered since the last report, busiest first,
/// and the counters behind it reset.
pub fn report(&self) -> Vec<String> {
let Ok(mut operations) = self.operations.lock() else {
return Vec::new();
};
let mut busy = operations
.iter_mut()
.filter(|entry| entry.1.answered > 0 || entry.1.hedged > 0)
.collect::<Vec<_>>();
busy.sort_by_key(|entry| Reverse(entry.1.answered));
let lines = busy
.into_iter()
.map(|(name, operation)| {
let mut line = String::new();
let _ = write!(
line,
"{name}: {} answered, p50 {} ms, p90 {} ms, p99 {} ms, hedge at {} ms, {} hedged, {} \
rescued",
operation.answered,
percentile(&operation.ring, 50),
percentile(&operation.ring, 90),
percentile(&operation.ring, 99),
operation.delay.unwrap_or(HEDGE_DEFAULT).as_millis(),
operation.hedged,
operation.rescued,
);
operation.answered = 0;
operation.hedged = 0;
operation.rescued = 0;
line
})
.collect();
drop(operations);
lines
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn the_delay_follows_the_slow_tail() {
let latency = Latency::default();
assert_eq!(latency.hedge_delay("abc/UserTweets"), HEDGE_DEFAULT);
for millis in 1..=100_u64 {
latency.record("abc/UserTweets", Duration::from_millis(millis * 10));
}
// Recomputed every few answers, so it can trail the newest ones. Two
// medians (about 960 ms) is later than the 95th percentile here.
let delay = latency.hedge_delay("other-id/UserTweets");
assert!(
(Duration::from_millis(900)..=Duration::from_millis(1000)).contains(&delay),
"{delay:?}"
);
}
#[test]
fn a_wide_tail_is_hedged_at_its_ninety_fifth_percentile() {
let latency = Latency::default();
for index in 0..96_u64 {
let millis = if index % 4 == 0 { 1000 } else { 200 };
latency.record("x/SearchTimeline", Duration::from_millis(millis));
}
assert_eq!(latency.hedge_delay("x/SearchTimeline"), Duration::from_millis(1000));
}
#[test]
fn the_delay_stays_inside_its_bounds() {
let latency = Latency::default();
for _ in 0..64 {
latency.record("x/UserByScreenName", Duration::from_millis(40));
latency.record("x/SearchTimeline", Duration::from_secs(9));
}
assert_eq!(latency.hedge_delay("x/UserByScreenName"), HEDGE_FLOOR);
assert_eq!(latency.hedge_delay("x/SearchTimeline"), HEDGE_CAP);
}
#[test]
fn a_report_resets_its_counters() {
let latency = Latency::default();
for _ in 0..40 {
latency.record("x/TweetDetail", Duration::from_millis(300));
}
latency.hedged("x/TweetDetail");
latency.rescued("x/TweetDetail");
let first = latency.report();
assert_eq!(first.len(), 1);
assert!(first[0].starts_with("TweetDetail: 40 answered"), "{}", first[0]);
assert!(first[0].ends_with("1 hedged, 1 rescued"), "{}", first[0]);
assert!(latency.report().is_empty());
}
}
src/api/mod.rs
+1
View File
@@ -5,6 +5,7 @@ mod client;
mod endpoints;
mod filters;
pub mod http;
mod latency;
mod limits_db;
mod operations;
mod parser;
src/api/proxy_pool.rs
+8 -3
View File
@@ -130,11 +130,16 @@ impl ProxyPool {
/// SOCKS5 config for this session, always the same port once pinned.
#[must_use]
pub fn for_session(&self, session_id: i64) -> ProxyConfig {
let port = self.pins.get(&session_id).copied().unwrap_or_else(|| {
self.at_port(self.port_for(session_id))
}
/// The port this session is pinned to.
#[must_use]
pub fn port_for(&self, session_id: i64) -> u16 {
self.pins.get(&session_id).copied().unwrap_or_else(|| {
let idx = (session_id.unsigned_abs() as usize) % self.endpoints.len();
self.endpoints[idx].port
});
self.at_port(port)
})
}
/// First endpoint, used for TID bootstrap and other unscoped API fetches.
src/main.rs
+50 -4
View File
@@ -12,7 +12,7 @@ mod views;
mod web_bot_auth;
use std::{
env,
env, io,
net::{IpAddr, SocketAddr},
sync::Arc,
time::Duration,
@@ -22,17 +22,23 @@ use axum::http::header;
use axum::{
Router, ServiceExt,
extract::Request,
http::{StatusCode, header::HeaderValue},
http::{Extensions, HeaderMap, StatusCode, Version, header::HeaderValue},
middleware,
};
use tokio::net::TcpListener;
use tower::{Layer as _, limit::ConcurrencyLimitLayer};
use tower_http::{
CompressionLevel,
compression::{
CompressionLayer,
predicate::{Predicate as _, SizeAbove},
},
normalize_path::NormalizePathLayer,
services::{ServeDir, ServeFile},
set_header::SetResponseHeaderLayer,
timeout::TimeoutLayer,
};
use tracing_appender::non_blocking::NonBlockingBuilder;
use tracing_subscriber::{fmt, layer::SubscriberExt as _, util::SubscriberInitExt as _};
use crate::{
@@ -56,15 +62,41 @@ pub struct AppState {
pub donations: Arc<donations::DonationLedger>,
}
/// Log lines held for stdout before new ones are dropped.
const LOG_BUFFER_LINES: usize = 10_000;
/// Pages, feeds, JSON, scripts and playlists. Images, video and audio come
/// compressed already, and a video relay must stream byte ranges as they are.
fn compressible(_: StatusCode, _: Version, headers: &HeaderMap, _: &Extensions) -> bool {
let Some(kind) = headers
.get(header::CONTENT_TYPE)
.and_then(|value| value.to_str().ok())
else {
return false;
};
let kind = kind.to_ascii_lowercase();
kind.starts_with("text/")
|| kind.contains("json")
|| kind.contains("xml")
|| kind.contains("javascript")
|| kind.contains("mpegurl")
}
#[tokio::main]
async fn main() -> eyre::Result<()> {
// Initialize logging
// pm2 reads stdout. While it lags, a direct write blocks the thread that
// is logging and every request scheduled on it; once this buffer is full,
// lines are dropped instead.
let (log_writer, _log_flush) = NonBlockingBuilder::default()
.buffered_lines_limit(LOG_BUFFER_LINES)
.lossy(true)
.finish(io::stdout());
tracing_subscriber::registry()
.with(
tracing_subscriber::EnvFilter::try_from_default_env()
.unwrap_or_else(|_| "teapot=debug,tower_http=debug".into()),
)
.with(fmt::layer())
.with(fmt::layer().with_writer(log_writer))
.init();
// Load configuration
@@ -113,6 +145,7 @@ async fn main() -> eyre::Result<()> {
let api = ApiClient::new(&config, sessions, proxies, fallback_proxies);
api.spawn_filter_sync();
api.spawn_search_check();
api.spawn_latency_report();
// Initialize GIF transcoder if local mode
let http_client = HttpClient::new(
@@ -223,6 +256,19 @@ async fn main() -> eyre::Result<()> {
HeaderValue::from_static("max-age=31536000; includeSubDomains"),
))
.layer(middleware::from_fn(routes::negotiate))
// Cloudflare compresses for visitors, but it reaches this process
// through a tunnel, and a page sent as-is crosses that at full size.
// Only gzip is built in. primp turns on tower-http's brotli and zstd
// decoders, and with those the layer picks `br` from Cloudflare's
// `gzip, br` and then sends the page uncompressed.
.layer(
CompressionLayer::new()
.no_br()
.no_zstd()
.no_deflate()
.quality(CompressionLevel::Fastest)
.compress_when(SizeAbove::new(1024).and(compressible)),
)
.layer(TimeoutLayer::with_status_code(
StatusCode::REQUEST_TIMEOUT,
Duration::from_secs(45),
src/routes/helpers.rs
+104 -2
View File
@@ -1,11 +1,18 @@
use std::{any::Any, collections::BTreeMap, future::Future, iter, slice, sync::LazyLock};
use std::{
any::Any,
collections::BTreeMap,
future::Future,
iter, slice,
sync::LazyLock,
time::Duration,
};
use axum::{
http::{StatusCode, header},
response::{Html, IntoResponse as _, Response},
};
use tokio::sync::Semaphore;
use tokio::{sync::Semaphore, time::timeout};
use crate::{
AppState,
@@ -22,6 +29,8 @@ use crate::{
/// Caps background profile prefetches so they cannot crowd out live pages.
static PROFILE_PREFETCH: LazyLock<Semaphore> = LazyLock::new(|| Semaphore::new(3));
const PREFETCH_PROFILES: usize = 4;
/// How long a page waits for its photo rail once everything else is in.
const RAIL_GRACE: Duration = Duration::from_millis(60);
#[derive(Clone)]
struct CachedRss {
@@ -237,6 +246,33 @@ pub async fn photo_rail(state: &AppState, username: &str, user_id: &str) -> Phot
}
}
/// A page's main fetch, and the photo rail if it lands by then or within
/// [`RAIL_GRACE`] after. The page is not held for a slow or stalled rail.
pub fn beside_rail<T, M, R>(main: M, rail: R) -> impl Future<Output = (T, PhotoRail)>
where
M: Future<Output = T>,
R: Future<Output = PhotoRail>,
{
// Boxed before the async block: two request futures held inline would
// make every page handler future that awaits this one twice as large.
let mut main = Box::pin(main);
let mut rail = Box::pin(rail);
async move {
let mut landed = None;
let value = loop {
tokio::select! {
value = &mut main => break value,
strip = &mut rail, if landed.is_none() => landed = Some(strip),
}
};
let strip = match landed {
Some(strip) => strip,
None => timeout(RAIL_GRACE, rail).await.unwrap_or_default(),
};
(value, strip)
}
}
/// Seed id mappings from tweets already on the page so a later profile click
/// can fetch tweets in the same wave as `UserByScreenName`.
pub fn remember_tweets(state: &AppState, tweets: &[Tweet]) {
@@ -314,6 +350,31 @@ pub async fn get_cached_user(state: &AppState, username: &str) -> Result<User> {
Ok(fetched)
}
/// The user and something that needs their numeric id, asked for in the same
/// wave when the id is already known. Without one, or when the name now
/// belongs to another account, `fetch` waits for `UserByScreenName`.
pub async fn with_user<T, F, Fut>(state: &AppState, username: &str, fetch: F) -> Result<(User, T)>
where
F: Fn(String) -> Fut,
Fut: Future<Output = T>,
{
let known = user_hint(state, username)
.map(|user| user.id)
.filter(|id| !id.is_empty());
let Some(id) = known else {
let user = get_cached_user(state, username).await?;
let fetched = fetch(user.id.clone()).await;
return Ok((user, fetched));
};
let (user, fetched) = tokio::join!(get_cached_user(state, username), fetch(id.clone()));
let user = user?;
if user.id == id {
return Ok((user, fetched));
}
let refetched = fetch(user.id.clone()).await;
Ok((user, refetched))
}
pub fn apply_context_to_user(user: &mut User, context: &AccountContext) {
user.account_based_in.clone_from(&context.account_based_in);
user
@@ -651,3 +712,44 @@ fn classify_error<'a>(err: &'a Error, generic: &'a str) -> (StatusCode, &'a str,
(status, title, message)
}
}
#[cfg(test)]
mod tests {
use std::time::Duration;
use tokio::time::sleep;
use super::beside_rail;
use crate::types::{GalleryPhoto, PhotoRail};
async fn rail_after(millis: u64) -> PhotoRail {
sleep(Duration::from_millis(millis)).await;
vec![GalleryPhoto {
url: "pic".into(),
tweet_id: "1".into(),
color: String::new(),
}]
}
async fn page_after(millis: u64) -> &'static str {
sleep(Duration::from_millis(millis)).await;
"page"
}
#[tokio::test]
async fn a_rail_that_lands_first_or_just_after_is_kept() {
let (page, rail) = beside_rail(page_after(40), rail_after(5)).await;
assert_eq!((page, rail.len()), ("page", 1));
let (_, rail) = beside_rail(page_after(20), rail_after(30)).await;
assert_eq!(rail.len(), 1);
}
#[tokio::test]
async fn the_page_does_not_wait_for_a_stalled_rail() {
let started = std::time::Instant::now();
let (page, rail) = beside_rail(page_after(10), rail_after(5_000)).await;
assert_eq!(page, "page");
assert!(rail.is_empty());
assert!(started.elapsed() < Duration::from_secs(1), "{:?}", started.elapsed());
}
}
src/routes/rss.rs
+17 -15
View File
@@ -8,7 +8,7 @@ use axum_extra::extract::CookieJar;
use serde::Deserialize;
use super::helpers::{
cache_rss, check_rss_cache, get_cached_user, rss_response, rss_response_with_min_id,
cache_rss, check_rss_cache, rss_response, rss_response_with_min_id, with_user,
};
use crate::{
AppState,
@@ -66,6 +66,7 @@ fn check_rss_enabled(state: &AppState) -> Result<()> {
}
/// Which user timeline variant to fetch for RSS.
#[derive(Clone, Copy)]
enum UserRssKind {
Tweets,
Replies,
@@ -117,20 +118,21 @@ async fn user_rss_handler(
return Ok(rss_response(rss, &tweets));
}
let user = get_cached_user(state, username).await?;
let timeline = match kind {
UserRssKind::Tweets => state.api.get_user_tweets(&user.id, cursor).await?,
UserRssKind::Replies => {
state
.api
.get_user_tweets_and_replies(&user.id, cursor)
.await?
},
UserRssKind::Reposts => state.api.get_user_reposts(&user.id, cursor).await?,
UserRssKind::Media => state.api.get_user_media(&user.id, cursor).await?,
};
let tweets = timeline.content.into_iter().flatten().collect::<Vec<_>>();
let (user, timeline) = with_user(state, username, |user_id| async move {
match kind {
UserRssKind::Tweets => state.api.get_user_tweets(&user_id, cursor).await,
UserRssKind::Replies => {
state
.api
.get_user_tweets_and_replies(&user_id, cursor)
.await
},
UserRssKind::Reposts => state.api.get_user_reposts(&user_id, cursor).await,
UserRssKind::Media => state.api.get_user_media(&user_id, cursor).await,
}
})
.await?;
let tweets = timeline?.content.into_iter().flatten().collect::<Vec<_>>();
let rss = rss_view::render_user_rss(&user, &tweets, &state.config, feed_kind, proxy);
if let Some(ref key) = rss_cache_key {
src/routes/timeline.rs
+110 -95
View File
@@ -12,7 +12,7 @@ use serde::Deserialize;
use super::{
helpers,
helpers::{extract_timeline, get_cached_user},
helpers::extract_timeline,
};
use crate::{
AppState,
@@ -222,25 +222,16 @@ async fn filtered_posts(
sort: &'static str,
is_scroll: bool,
) -> Result<Response> {
let user = get_cached_user(state, username).await?;
let popular = sort == "popular";
let cache_key = cache_keys::filtered_posts(username, filter, sort, cursor);
let timeline_result = if let Some(Hit::Fresh(hit) | Hit::Stale(hit)) =
state.cache.lookup::<Timeline>(&cache_key)
{
Ok(hit)
} else {
let result = state
.api
.get_filtered_posts(&user.id, cursor, filter, popular)
.await;
if let Ok(ref data) = result {
state
.cache
.set_swr(&cache_key, data, ttl::DEFAULT, ttl::DEFAULT_STALE);
}
result
};
let (user, (timeline_result, photo_rail)) =
helpers::with_user(state, username, |user_id| async move {
let posts = filtered_timeline(state, username, &user_id, cursor, filter, sort);
if is_scroll {
(posts.await, Vec::new())
} else {
helpers::beside_rail(posts, helpers::photo_rail(state, username, &user_id)).await
}
})
.await?;
match timeline_result {
Ok(data) => {
@@ -261,10 +252,7 @@ async fn filtered_posts(
);
return Ok(Html(content.into_string()).into_response());
}
let (photo_rail, show_articles) = tokio::join!(
helpers::photo_rail(state, username, &user.id),
helpers::user_has_articles(state, &user.id),
);
let show_articles = helpers::user_has_articles(state, &user.id).await;
let newer = cursor.is_some().then_some(base_url.as_str());
let timeline_content = html! {
(timeline::render_timeline_with_prefs(
@@ -307,6 +295,31 @@ async fn filtered_posts(
}
}
/// One page of the posts menu, from cache when it has one.
async fn filtered_timeline(
state: &AppState,
username: &str,
user_id: &str,
cursor: Option<&str>,
filter: &'static str,
sort: &'static str,
) -> Result<Timeline> {
let cache_key = cache_keys::filtered_posts(username, filter, sort, cursor);
if let Some(Hit::Fresh(hit) | Hit::Stale(hit)) = state.cache.lookup::<Timeline>(&cache_key) {
return Ok(hit);
}
let result = state
.api
.get_filtered_posts(user_id, cursor, filter, sort == "popular")
.await;
if let Ok(ref data) = result {
state
.cache
.set_swr(&cache_key, data, ttl::DEFAULT, ttl::DEFAULT_STALE);
}
result
}
async fn user_timeline(
State(state): State<AppState>,
jar: CookieJar,
@@ -451,8 +464,6 @@ async fn user_tab_handler(
include_photo_rail: bool,
view: &str,
) -> Result<Response> {
let user = get_cached_user(state, username).await?;
let (cache_kind, tab_str, title_prefix) = match tab {
TimelineKind::Replies => ("replies", "with_replies", "Tweets & replies from"),
TimelineKind::Reposts => ("reposts", "reposts", "Reposts from"),
@@ -461,70 +472,16 @@ async fn user_tab_handler(
_ => unreachable!(),
};
// Fetch timeline (with cache for first page)
let fetch_timeline = async {
let cache_key = cache_keys::timeline(username, cache_kind);
if cursor.is_none() {
if let Some(cached) = helpers::swr_take::<Timeline, _, _>(state, &cache_key, {
let user_id = user.id.clone();
let cache_key = cache_key.clone();
move |state| async move {
let result = match tab {
TimelineKind::Replies => {
state.api.get_user_tweets_and_replies(&user_id, None).await
},
TimelineKind::Reposts => state.api.get_user_reposts(&user_id, None).await,
TimelineKind::Articles => state.api.get_user_articles(&user_id, None).await,
TimelineKind::Media => state.api.get_user_media(&user_id, None).await,
_ => return,
};
if let Ok(data) = result {
state
.cache
.set_swr(&cache_key, &data, ttl::DEFAULT, ttl::DEFAULT_STALE);
}
}
}) {
return Ok(cached);
let (user, (timeline_result, photo_rail)) =
helpers::with_user(state, username, |user_id| async move {
let timeline = tab_timeline(state, username, &user_id, cursor, tab, cache_kind);
if include_photo_rail {
helpers::beside_rail(timeline, helpers::photo_rail(state, username, &user_id)).await
} else {
(timeline.await, Vec::new())
}
let result = match tab {
TimelineKind::Replies => state.api.get_user_tweets_and_replies(&user.id, None).await,
TimelineKind::Reposts => state.api.get_user_reposts(&user.id, None).await,
TimelineKind::Articles => state.api.get_user_articles(&user.id, None).await,
TimelineKind::Media => state.api.get_user_media(&user.id, None).await,
_ => unreachable!(),
};
if let Ok(ref data) = result {
state
.cache
.set_swr(&cache_key, data, ttl::DEFAULT, ttl::DEFAULT_STALE);
}
result
} else {
match tab {
TimelineKind::Replies => {
state
.api
.get_user_tweets_and_replies(&user.id, cursor)
.await
},
TimelineKind::Reposts => state.api.get_user_reposts(&user.id, cursor).await,
TimelineKind::Articles => state.api.get_user_articles(&user.id, cursor).await,
TimelineKind::Media => state.api.get_user_media(&user.id, cursor).await,
_ => unreachable!(),
}
}
};
let (timeline_result, photo_rail) = if include_photo_rail {
let (tl, rail) = tokio::join!(
fetch_timeline,
helpers::photo_rail(state, username, &user.id)
);
(tl, rail)
} else {
(fetch_timeline.await, vec![])
};
})
.await?;
let has_request_cursor = cursor.is_some();
@@ -605,6 +562,56 @@ async fn user_tab_handler(
}
}
/// A profile tab's timeline, with the cache for its first page.
async fn tab_timeline(
state: &AppState,
username: &str,
user_id: &str,
cursor: Option<&str>,
tab: TimelineKind,
cache_kind: &str,
) -> Result<Timeline> {
if cursor.is_some() {
return fetch_tab(state, user_id, cursor, tab).await;
}
let cache_key = cache_keys::timeline(username, cache_kind);
if let Some(cached) = helpers::swr_take::<Timeline, _, _>(state, &cache_key, {
let user_id = user_id.to_owned();
let cache_key = cache_key.clone();
move |state| async move {
if let Ok(data) = fetch_tab(&state, &user_id, None, tab).await {
state
.cache
.set_swr(&cache_key, &data, ttl::DEFAULT, ttl::DEFAULT_STALE);
}
}
}) {
return Ok(cached);
}
let result = fetch_tab(state, user_id, None, tab).await;
if let Ok(ref data) = result {
state
.cache
.set_swr(&cache_key, data, ttl::DEFAULT, ttl::DEFAULT_STALE);
}
result
}
async fn fetch_tab(
state: &AppState,
user_id: &str,
cursor: Option<&str>,
tab: TimelineKind,
) -> Result<Timeline> {
match tab {
TimelineKind::Replies => state.api.get_user_tweets_and_replies(user_id, cursor).await,
TimelineKind::Reposts => state.api.get_user_reposts(user_id, cursor).await,
TimelineKind::Articles => state.api.get_user_articles(user_id, cursor).await,
TimelineKind::Media => state.api.get_user_media(user_id, cursor).await,
_ => Err(Error::InvalidUrl("Invalid tab parameter".into())),
}
}
async fn user_replies(
State(state): State<AppState>,
jar: CookieJar,
@@ -696,7 +703,6 @@ async fn user_search(
Query(query): Query<UserSearchQuery>,
) -> Result<Response> {
let prefs = Prefs::from_cookies(&jar, &state.config);
let user = get_cached_user(&state, &username).await?;
let search_query = query.query.as_deref().unwrap_or("");
let search_action = format!("/{username}/search");
@@ -707,12 +713,17 @@ async fn user_search(
} else {
format!("from:{username} {search_query}")
};
let (photo_rail, search_result) = tokio::join!(
helpers::photo_rail(&state, &username, &user.id),
// The search names the account, not its id, so it never waits on the user.
let (state_ref, name) = (&state, username.as_str());
let (fetched, search_result) = tokio::join!(
helpers::with_user(state_ref, name, |user_id| async move {
helpers::photo_rail(state_ref, name, &user_id).await
}),
state
.api
.search(&api_query, query.cursor.as_deref(), "Latest"),
);
let (user, photo_rail) = fetched?;
match search_result {
Ok(results) => {
@@ -818,9 +829,13 @@ async fn multi_user_timeline(
// A spawned task starts with an empty task-local scope, so the caller
// has to be carried over or these fan-out calls go unbilled.
tokio::spawn(budget::scoped(async move {
let user = get_cached_user(&state, &username).await.ok()?;
let timeline = state.api.get_user_tweets(&user.id, None).await.ok()?;
Some(timeline.content.into_iter().flatten().collect::<Vec<_>>())
let api = &state.api;
let (_, timeline) = helpers::with_user(&state, &username, |user_id| async move {
api.get_user_tweets(&user_id, None).await
})
.await
.ok()?;
Some(timeline.ok()?.content.into_iter().flatten().collect::<Vec<_>>())
}))
})
.collect();