Stop a bad span or a dead connection from stalling the page.

X sometimes sends a negative text index and drops HTTP/2 streams; those used to fail the whole response or sit until the gateway gave up, and the small concurrency caps made every other call wait.

Co-authored-by: Cursor <[email protected]>

src/api/client.rs

@@ -38,6 +38,13 @@ const GRAPHQL_REPLACEMENTS: usize = 2;
 /// not walk the pool — that used to zero every cookie for 15 minutes.
 const RATE_LIMIT_REPLACEMENTS: usize = 5;
 
+/// Edge 502/503/504, including an empty body and a stream the proxy reset.
+fn is_gateway_blip(message: &str) -> bool {
+   message.starts_with("Status 502")
+      || message.starts_with("Status 503")
+      || message.starts_with("Status 504")
+}
+
 fn is_denylisted(message: &str) -> bool {
    message.contains("QueryDenylistedFailure") || message.contains("Query is denylisted")
 }
@@ -462,6 +469,19 @@ impl ApiClient {
             Err(Error::TransientUpstream) => {
                replacement == 0 && session.kind == SessionKind::Cookie
             },
+            // A 502/503/504 is the edge dropping the stream, not this account's
+            // quota. Try one other session. Do not mark the first as refused:
+            // that path is for an account X answers with an empty 404.
+            Err(Error::TwitterApi(message)) if is_gateway_blip(message) && replacement == 0 => {
+               true
+            },
+            // Headers never arrived, so the pooled connection is the thing
+            // that stalled. Another account is pinned to another exit.
+            Err(Error::Internal(message))
+               if message == "API headers stalled" && replacement == 0 =>
+            {
+               true
+            },
             Err(Error::RateLimited) if replacement < RATE_LIMIT_REPLACEMENTS => {
                self
                   .sessions
@@ -878,7 +898,16 @@ impl ApiClient {
             return Err(Error::NotFound("Not found".into()));
          }
 
-         if status.is_server_error() {
+         if matches!(status.as_u16(), 502 | 503 | 504) {
+            tracing::warn!(
+               session_id = session.id,
+               session_user = %session.username,
+               "API request failed: {status} - {body}"
+            );
+            // The gateway reset this stream. Drop the HTTP/2 session so the
+            // next call from this account does not wait on it again.
+            self.client.retire(Some(&self.egress_for(session)));
+         } else if status.is_server_error() {
             tracing::error!(
                session_id = session.id,
                session_user = %session.username,

src/api/http.rs

@@ -39,12 +39,19 @@ const BODY_IDLE_TIMEOUT: Duration = Duration::from_secs(20);
 /// Idle connections are kept this long. A session's connection to X is
 /// multiplexed HTTP/2 and pinged, so a dead one is noticed rather than hung on.
 const POOL_IDLE: Duration = Duration::from_secs(300);
-const POOL_PER_HOST: usize = 4;
+/// Idle connections kept per account. High enough that in-flight calls are not
+/// stuck sharing a handful of sockets.
+const POOL_PER_HOST: usize = 256;
 /// CDN media normally answers in tens of milliseconds. A pooled connection the
 /// path has already forgotten sits in TCP retransmits for several seconds and
 /// then delivers, which is the video that "loads" after a long pause. Give up
 /// on headers by then and open another connection.
 const MEDIA_HEADER_DEADLINE: Duration = Duration::from_secs(2);
+/// X answers a healthy GraphQL call in well under a second. A pooled HTTP/2
+/// connection the proxy has already dropped sits until the gateway gives up
+/// (about five seconds) and then returns 503 "remote refused stream reset".
+/// Abandon headers by now and open another connection.
+const API_HEADER_DEADLINE: Duration = Duration::from_secs(1);
 const KEEP_ALIVE_INTERVAL: Duration = Duration::from_secs(45);
 const KEEP_ALIVE_TIMEOUT: Duration = Duration::from_secs(15);
 /// After an HTML 429, skip the direct address this long and send every
@@ -286,10 +293,7 @@ impl HttpClient {
    /// time, so those clients belong to the browser rather than the account:
    /// every account presenting that browser shares them. Kept per account,
    /// each one would gather a client per exit and reuse almost none.
-   fn client_for(&self, via: Option<&Egress>) -> Result<primp::Client> {
-      let Some(egress) = via else {
-         return Ok(self.default.clone());
-      };
+   fn client_key(&self, egress: &Egress) -> (i64, u64) {
       let identity = browser::identity_for(egress.session);
       let tag = proxy_tag(egress.proxy.as_ref());
       let pooled = self
@@ -301,7 +305,27 @@ impl HttpClient {
       } else {
          egress.session
       };
-      let key = (owner, tag);
+      (owner, tag)
+   }
+
+   /// Drop the pooled connection for this account so the next call dials a
+   /// new one. A wedged HTTP/2 session otherwise accepts the next stream and
+   /// stalls it until the proxy resets it.
+   pub(crate) fn retire(&self, via: Option<&Egress>) {
+      let Some(egress) = via else {
+         return;
+      };
+      let key = self.client_key(egress);
+      if let Ok(mut clients) = self.sessions.lock() {
+         clients.remove(&key);
+      }
+   }
+
+   fn client_for(&self, via: Option<&Egress>) -> Result<primp::Client> {
+      let Some(egress) = via else {
+         return Ok(self.default.clone());
+      };
+      let key = self.client_key(egress);
       if let Some(client) = self
          .sessions
          .lock()
@@ -316,6 +340,7 @@ impl HttpClient {
       } else {
          POOL_PER_HOST
       };
+      let identity = browser::identity_for(egress.session);
       let client = build_client(
          identity,
          egress.proxy.as_ref(),
@@ -495,6 +520,35 @@ impl HttpClient {
          return Ok(Response { inner });
       }
 
+      if self.purpose == Purpose::Api && is_idempotent(method) {
+         let inner = match tokio::time::timeout(API_HEADER_DEADLINE, send()).await {
+            Ok(Ok(inner)) => inner,
+            Ok(Err(err)) if is_transport_blip(&err) || err.is_connect() => {
+               tracing::warn!(uri, error = %err, "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");
+               match tokio::time::timeout(
+                  API_HEADER_DEADLINE,
+                  self.fresh_headers(method, uri, extra_headers, body, via),
+               )
+               .await
+               {
+                  Ok(Ok(inner)) => inner,
+                  Ok(Err(err)) => return Err(err),
+                  Err(_elapsed) => {
+                     self.retire(via);
+                     tracing::warn!(uri, "API headers stalled again");
+                     return Err(Error::Internal("API headers stalled".into()));
+                  },
+               }
+            },
+            Ok(Err(err)) => return Err(describe(err)),
+         };
+         return Ok(Response { inner });
+      }
+
       let inner = match send().await {
          Ok(inner) => inner,
          // A pooled connection the far end closed a moment ago fails on the
@@ -509,6 +563,26 @@ impl HttpClient {
       };
       Ok(Response { inner })
    }
+
+   /// One more attempt on a connection that is not the one that just failed.
+   async fn fresh_headers(
+      &self,
+      method: &Method,
+      uri: &str,
+      extra_headers: &HeaderMap,
+      body: &Bytes,
+      via: Option<&Egress>,
+   ) -> Result<primp::Response> {
+      self.retire(via);
+      let client = self.client_for(via)?;
+      let mut request = client
+         .request(method.clone(), uri)
+         .headers(extra_headers.clone());
+      if !body.is_empty() {
+         request = request.body(body.clone());
+      }
+      request.send().await.map_err(describe)
+   }
 }
 
 /// The browser presented when no account is involved.

src/api/schema.rs

@@ -88,7 +88,9 @@ pub struct BirdwatchSubtitle {
 #[derive(Deserialize, Default)]
 #[serde(rename_all = "camelCase")]
 pub struct BirdwatchEntity {
+   #[serde(default, deserialize_with = "deser_opt_index")]
    pub from_index: Option<u64>,
+   #[serde(default, deserialize_with = "deser_opt_index")]
    pub to_index: Option<u64>,
    #[serde(rename = "ref")]
    pub ref_data: Option<BirdwatchRef>,
@@ -299,7 +301,7 @@ pub struct Entities {
 
 #[derive(Deserialize, Default)]
 pub struct UrlEntity {
-   #[serde(default)]
+   #[serde(default, deserialize_with = "deser_index_list")]
    pub indices: Vec<u64>,
    pub url: Option<String>,
    pub expanded_url: Option<String>,
@@ -307,14 +309,14 @@ pub struct UrlEntity {
 
 #[derive(Deserialize, Default)]
 pub struct HashtagEntity {
-   #[serde(default)]
+   #[serde(default, deserialize_with = "deser_index_list")]
    pub indices: Vec<u64>,
    pub text: Option<String>,
 }
 
 #[derive(Deserialize, Default)]
 pub struct MentionEntity {
-   #[serde(default)]
+   #[serde(default, deserialize_with = "deser_index_list")]
    pub indices: Vec<u64>,
    pub screen_name: Option<String>,
    pub name: Option<String>,
@@ -322,7 +324,7 @@ pub struct MentionEntity {
 
 #[derive(Deserialize, Default)]
 pub struct SymbolEntity {
-   #[serde(default)]
+   #[serde(default, deserialize_with = "deser_index_list")]
    pub indices: Vec<u64>,
    pub text: Option<String>,
 }
@@ -534,6 +536,7 @@ pub struct TweetLegacy {
    pub conversation_id_str: Option<String>,
    pub self_thread: Option<SelfThread>,
    pub retweeted_status_result: Option<NestedResult<TweetData>>,
+   #[serde(default, deserialize_with = "deser_opt_index_list")]
    pub display_text_range: Option<Vec<u64>>,
    pub place: Option<Place>,
    pub withheld_in_countries: Option<Vec<String>>,
@@ -939,6 +942,37 @@ where
       .unwrap_or_default())
 }
 
+/// X sometimes sends `-1` for an unknown span. That is not a position.
+fn index_floor(value: i64) -> u64 {
+   u64::try_from(value).unwrap_or(0)
+}
+
+fn deser_index_list<'de, D>(deserializer: D) -> Result<Vec<u64>, D::Error>
+where
+   D: Deserializer<'de>,
+{
+   Ok(Vec::<i64>::deserialize(deserializer)?
+      .into_iter()
+      .map(index_floor)
+      .collect())
+}
+
+fn deser_opt_index_list<'de, D>(deserializer: D) -> Result<Option<Vec<u64>>, D::Error>
+where
+   D: Deserializer<'de>,
+{
+   Ok(Option::<Vec<i64>>::deserialize(deserializer)?.map(|values| {
+      values.into_iter().map(index_floor).collect()
+   }))
+}
+
+fn deser_opt_index<'de, D>(deserializer: D) -> Result<Option<u64>, D::Error>
+where
+   D: Deserializer<'de>,
+{
+   Ok(Option::<i64>::deserialize(deserializer)?.map(index_floor))
+}
+
 /// Extract (start, end) from a raw indices array.
 #[expect(
    clippy::cast_possible_truncation,
@@ -951,6 +985,21 @@ pub fn indices(raw: &[u64]) -> (usize, usize) {
    )
 }
 
+#[cfg(test)]
+mod index_tests {
+   use super::*;
+
+   #[test]
+   fn negative_spans_do_not_reject_the_tweet() {
+      let legacy: TweetLegacy = serde_json::from_str(
+         r#"{"display_text_range":[-1,12],"entities":{"urls":[{"indices":[-5,4],"url":"https://t.co/abc"}]}}"#,
+      )
+      .unwrap();
+      assert_eq!(legacy.display_text_range, Some(vec![0, 12]));
+      assert_eq!(legacy.entities.unwrap().urls[0].indices, vec![0, 4]);
+   }
+}
+
 #[path = "schema_endpoints.rs"]
 mod endpoint_types;
 pub use endpoint_types::*;

src/routes/timeline.rs

@@ -74,16 +74,14 @@ async fn refresh_profile(state: &AppState, username: &str) {
 async fn fetch_profile(state: &AppState, username: &str) -> Result<Profile> {
    let hint = helpers::user_hint(state, username);
    let rail = helpers::cached_photo_rail(state, username);
-   let (profile_res, about) = tokio::join!(
-      state
-         .api
-         .get_profile_hinted(username, None, hint.as_ref(), rail),
-      helpers::load_account_context(state, username),
-   );
-   let mut profile = profile_res?;
-   if let Some(context) = about {
-      helpers::apply_context_to_user(&mut profile.user, &context);
-   }
+   let mut profile = state
+      .api
+      .get_profile_hinted(username, None, hint.as_ref(), rail)
+      .await?;
+   // About Account is a separate X call and is the one that sits on a dead
+   // connection. The timeline does not need it; a cached copy is applied and
+   // a miss is filled in the background.
+   helpers::apply_cached_account_context(state, &mut profile.user);
    helpers::store_profile(state, &profile);
    Ok(profile)
 }

src/transcode.rs

@@ -41,7 +41,7 @@ impl GifTranscoder {
          cache,
          http_client,
          cache_dir: PathBuf::from(config.cache_dir),
-         semaphore: Semaphore::new(3),
+         semaphore: Semaphore::new(64),
          inflight: Mutex::new(HashMap::new()),
       })
    }

src/translation.rs

@@ -9,7 +9,7 @@ use crate::error::{Error, Result};
 
 const MAX_TRANSLATIONS_PER_WINDOW: usize = 60;
 const WINDOW: Duration = Duration::from_secs(60);
-const MAX_CONCURRENT_TRANSLATIONS: usize = 2;
+const MAX_CONCURRENT_TRANSLATIONS: usize = 1024;
 
 /// Global guard for translation backends.
 pub struct TranslationLimiter {