Write logs from a background thread so a lagging pm2 drops lines instead of stalling requests, and log GraphQL calls without the feature blob.

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

Cargo.lock

@@ -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

@@ -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/http.rs

@@ -719,11 +719,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)),
@@ -735,11 +742,18 @@ impl HttpClient {
          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");
+               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");
+               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,7 +764,7 @@ 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()));
                   },
                }
@@ -1054,6 +1068,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 {
@@ -1199,6 +1220,17 @@ fn proxy_on(port: u16) -> ProxyConfig {
 mod tests {
    use super::*;
 
+   #[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 fallback_pool_uses_a_different_exit_each_call() {
       let proxies = vec![proxy_on(10001), proxy_on(10002), proxy_on(10003)];

src/main.rs

@@ -12,7 +12,7 @@ mod views;
 mod web_bot_auth;
 
 use std::{
-   env,
+   env, io,
    net::{IpAddr, SocketAddr},
    sync::Arc,
    time::Duration,
@@ -33,6 +33,7 @@ use tower_http::{
    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 +57,24 @@ 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;
+
 #[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