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

Undo parking slow proxy exits, copying long X calls, and fetching a tab's user and posts together.

Browse Source 
Under load these let requests to teapot pile up (31k open, 17 GB) until every X call took about 8 s, and xitter.cf timed out. This reverts commit 0988cae.

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

No files matched your search

Cargo.lock
Generated
-35
View File
@@ -394,21 +394,6 @@ 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"
@@ -1771,12 +1756,6 @@ 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"
@@ -1850,7 +1829,6 @@ dependencies = [
"tower",
"tower-http 0.7.0",
"tracing",
"tracing-appender",
"tracing-subscriber",
"xitter-txid",
]
@@ -2154,19 +2132,6 @@ 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,7 +71,6 @@ 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
+54 -96
View File
@@ -4,7 +4,6 @@ use std::{
fmt::Write as _,
ops::Deref,
path::Path,
result::Result as StdResult,
sync::{
Arc,
atomic::{AtomicBool, AtomicUsize, Ordering},
@@ -94,16 +93,6 @@ 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>,
@@ -415,14 +404,61 @@ impl SessionPool {
}
loop {
let start = self.cursor.fetch_add(1, Ordering::Relaxed);
let mut limits = self.limits.write().await;
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 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);
}
let waiter = self.freed.notified();
tokio::pin!(waiter);
@@ -434,84 +470,6 @@ 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
+4 -159
View File
@@ -1,21 +1,14 @@
use std::{
sync::Arc,
time::{Duration, Instant},
};
use std::sync::Arc;
use axum::http::header;
use serde::{Deserialize, de::DeserializeOwned};
use tokio::{
sync::Semaphore,
time::{interval, sleep, timeout},
};
use tokio::time::timeout;
use super::{
ProxyPool, SessionLease, SessionPool, TidClient,
budget::{self, ClientBudget},
endpoints,
http::{Egress, HttpClient, ProxyConfig, Purpose},
latency::Latency,
operations::Operations,
parser,
};
@@ -63,28 +56,6 @@ 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 {
@@ -319,8 +290,6 @@ pub struct ApiClient {
budget: ClientBudget,
tid_enabled: bool,
denylist: super::search_denylist::SearchDenylist,
latency: Arc<Latency>,
hedges: Arc<Semaphore>,
}
impl ApiClient {
@@ -369,39 +338,9 @@ 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 {
@@ -540,10 +479,9 @@ impl ApiClient {
where
T: DeserializeOwned,
{
let (answered_on, mut last) = self
.hedged_attempt(session, endpoint, variables, features, field_toggles, retry_kind)
let mut last = self
.graphql_request_inner(&session, endpoint, variables, features, field_toggles)
.await;
session = answered_on;
// Quota failures may walk RATE_LIMIT_REPLACEMENTS accounts. The other
// cases stay capped at GRAPHQL_REPLACEMENTS inside the match.
@@ -639,99 +577,6 @@ 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
+15 -355
View File
@@ -18,7 +18,7 @@ use std::{
atomic::{AtomicU64, AtomicUsize, Ordering},
},
task::{Context, Poll},
time::{Duration, Instant, SystemTime, UNIX_EPOCH},
time::{Duration, SystemTime, UNIX_EPOCH},
};
use axum::http::{HeaderMap, HeaderName, HeaderValue, Method, StatusCode, header};
@@ -63,27 +63,6 @@ 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 {
@@ -175,158 +154,16 @@ 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 or answers it far slower than the other exits.
/// its public IP.
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)))
}
@@ -497,35 +334,15 @@ impl HttpClient {
cooldown_secs = EXIT_BLOCK_FOR.as_secs(),
"proxy exit circuit breaker ready"
);
self.primary_pool = Some(Arc::new(PrimaryPool::new(proxies, tags)));
self.primary_pool = Some(Arc::new(PrimaryPool {
proxies,
tags,
blocked_until: Mutex::new(HashMap::new()),
}));
}
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()
}
@@ -902,18 +719,11 @@ 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 = loggable(uri),
error = %describe(err.without_url()),
"media fetch failed, retrying"
);
tracing::warn!(uri, error = %err, "media fetch failed, retrying");
send().await.map_err(describe)?
},
Err(_elapsed) => {
tracing::warn!(
uri = loggable(uri),
"media headers stalled, opening a new connection"
);
tracing::warn!(uri, "media headers stalled, opening a new connection");
send().await.map_err(describe)?
},
Ok(Err(err)) => return Err(describe(err)),
@@ -922,27 +732,14 @@ 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)) => {
probe.answered();
inner
},
Ok(Ok(inner)) => inner,
Ok(Err(err)) if is_transport_blip(&err) || err.is_connect() => {
probe.failed();
tracing::warn!(
uri = loggable(uri),
error = %describe(err.without_url()),
"API connection failed, opening another"
);
tracing::warn!(uri, error = %err, "API connection failed, opening another");
self.fresh_headers(method, uri, extra_headers, body, via).await?
},
Err(_elapsed) => {
probe.failed();
tracing::warn!(
uri = loggable(uri),
"API headers stalled, opening another connection"
);
tracing::warn!(uri, "API headers stalled, opening another connection");
match tokio::time::timeout(
API_HEADER_DEADLINE,
self.fresh_headers(method, uri, extra_headers, body, via),
@@ -953,15 +750,12 @@ impl HttpClient {
Ok(Err(err)) => return Err(err),
Err(_elapsed) => {
self.retire(via);
tracing::warn!(uri = loggable(uri), "API headers stalled again");
tracing::warn!(uri, "API headers stalled again");
return Err(Error::Internal("API headers stalled".into()));
},
}
},
Ok(Err(err)) => {
probe.failed();
return Err(describe(err));
},
Ok(Err(err)) => return Err(describe(err)),
};
return Ok(Response { inner });
}
@@ -998,60 +792,7 @@ impl HttpClient {
if !body.is_empty() {
request = request.body(body.clone());
}
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);
}
request.send().await.map_err(describe)
}
}
@@ -1313,13 +1054,6 @@ 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 {
@@ -1510,80 +1244,6 @@ 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
@@ -1,224 +0,0 @@
//! 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,7 +5,6 @@ mod client;
mod endpoints;
mod filters;
pub mod http;
mod latency;
mod limits_db;
mod operations;
mod parser;
src/api/proxy_pool.rs
+3 -8
View File
@@ -130,16 +130,11 @@ impl ProxyPool {
/// SOCKS5 config for this session, always the same port once pinned.
#[must_use]
pub fn for_session(&self, session_id: i64) -> ProxyConfig {
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 port = 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
+4 -50
View File
@@ -12,7 +12,7 @@ mod views;
mod web_bot_auth;
use std::{
env, io,
env,
net::{IpAddr, SocketAddr},
sync::Arc,
time::Duration,
@@ -22,23 +22,17 @@ use axum::http::header;
use axum::{
Router, ServiceExt,
extract::Request,
http::{Extensions, HeaderMap, StatusCode, Version, header::HeaderValue},
http::{StatusCode, 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::{
@@ -62,41 +56,15 @@ 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<()> {
// 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());
// Initialize logging
tracing_subscriber::registry()
.with(
tracing_subscriber::EnvFilter::try_from_default_env()
.unwrap_or_else(|_| "teapot=debug,tower_http=debug".into()),
)
.with(fmt::layer().with_writer(log_writer))
.with(fmt::layer())
.init();
// Load configuration
@@ -145,7 +113,6 @@ 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(
@@ -256,19 +223,6 @@ 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
+2 -104
View File
@@ -1,18 +1,11 @@
use std::{
any::Any,
collections::BTreeMap,
future::Future,
iter, slice,
sync::LazyLock,
time::Duration,
};
use std::{any::Any, collections::BTreeMap, future::Future, iter, slice, sync::LazyLock};
use axum::{
http::{StatusCode, header},
response::{Html, IntoResponse as _, Response},
};
use tokio::{sync::Semaphore, time::timeout};
use tokio::sync::Semaphore;
use crate::{
AppState,
@@ -29,8 +22,6 @@ 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 {
@@ -246,33 +237,6 @@ 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]) {
@@ -350,31 +314,6 @@ 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
@@ -712,44 +651,3 @@ 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
+15 -17
View File
@@ -8,7 +8,7 @@ use axum_extra::extract::CookieJar;
use serde::Deserialize;
use super::helpers::{
cache_rss, check_rss_cache, rss_response, rss_response_with_min_id, with_user,
cache_rss, check_rss_cache, get_cached_user, rss_response, rss_response_with_min_id,
};
use crate::{
AppState,
@@ -66,7 +66,6 @@ fn check_rss_enabled(state: &AppState) -> Result<()> {
}
/// Which user timeline variant to fetch for RSS.
#[derive(Clone, Copy)]
enum UserRssKind {
Tweets,
Replies,
@@ -118,21 +117,20 @@ async fn user_rss_handler(
return Ok(rss_response(rss, &tweets));
}
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 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 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
+95 -110
View File
@@ -12,7 +12,7 @@ use serde::Deserialize;
use super::{
helpers,
helpers::extract_timeline,
helpers::{extract_timeline, get_cached_user},
};
use crate::{
AppState,
@@ -222,16 +222,25 @@ async fn filtered_posts(
sort: &'static str,
is_scroll: bool,
) -> Result<Response> {
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?;
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
};
match timeline_result {
Ok(data) => {
@@ -252,7 +261,10 @@ async fn filtered_posts(
);
return Ok(Html(content.into_string()).into_response());
}
let show_articles = helpers::user_has_articles(state, &user.id).await;
let (photo_rail, show_articles) = tokio::join!(
helpers::photo_rail(state, username, &user.id),
helpers::user_has_articles(state, &user.id),
);
let newer = cursor.is_some().then_some(base_url.as_str());
let timeline_content = html! {
(timeline::render_timeline_with_prefs(
@@ -295,31 +307,6 @@ 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,
@@ -464,6 +451,8 @@ 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"),
@@ -472,16 +461,70 @@ async fn user_tab_handler(
_ => unreachable!(),
};
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())
// 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);
}
})
.await?;
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![])
};
let has_request_cursor = cursor.is_some();
@@ -562,56 +605,6 @@ 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,
@@ -703,6 +696,7 @@ 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");
@@ -713,17 +707,12 @@ async fn user_search(
} else {
format!("from:{username} {search_query}")
};
// 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
}),
let (photo_rail, search_result) = tokio::join!(
helpers::photo_rail(&state, &username, &user.id),
state
.api
.search(&api_query, query.cursor.as_deref(), "Latest"),
);
let (user, photo_rail) = fetched?;
match search_result {
Ok(results) => {
@@ -829,13 +818,9 @@ 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 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<_>>())
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<_>>())
}))
})
.collect();