From 7e41876c2d1039351fa2638e40b4be9ee53773fb Mon Sep 17 00:00:00 2001
From: dawn <90008@klbr.net>
Date: Sat, 3 Oct 2026 16:11:49 +0300
Subject: [PATCH] [ingest] share one tls connector across firehose connections

tokio-websockets builds a connector for every connect that doesn't get one,
loading and parsing the system root store each time. with ~6.6k PDS
sources reconnecting after a restart that pegged the runtime and kept
/_health unanswered for minutes. build it once and hand it to every wss
connect.
---
 src/ingest/stream.rs | 16 ++++++++++++++--
 1 file changed, 14 insertions(+), 2 deletions(-)

diff --git a/src/ingest/stream.rs b/src/ingest/stream.rs
index 2260aff..540d9d1 100644
--- a/src/ingest/stream.rs
+++ b/src/ingest/stream.rs
@@ -1,4 +1,5 @@
 use std::convert::Infallible;
+use std::sync::LazyLock;
 use std::time::Duration;
 
 use axum::http::Uri;
@@ -8,7 +9,7 @@ use jacquard_common::error::DecodeError;
 use miette::Diagnostic;
 use smol_str::format_smolstr;
 use thiserror::Error;
-use tokio_websockets::{ClientBuilder, Message as WsMsg, WebSocketStream};
+use tokio_websockets::{ClientBuilder, Connector, Message as WsMsg, WebSocketStream};
 
 use crate::net::{PublicEndpoint, PublicResolver};
 use tracing::trace;
@@ -66,6 +67,12 @@ impl From<serde_ipld_dagcbor::DecodeError<Infallible>> for FirehoseError {
     }
 }
 
+/// shared by every firehose connection: without one, tokio-websockets builds a
+/// connector per connect, which re-reads and parses the system root store each
+/// time and starves the runtime when thousands of PDS streams start at once.
+/// if it can't be built, connects fall back to that path and report its error.
+static TLS_CONNECTOR: LazyLock<Option<Connector>> = LazyLock::new(|| Connector::new().ok());
+
 pub struct FirehoseStream {
     ws: WebSocketStream<tokio_websockets::MaybeTlsStream<tokio::net::TcpStream>>,
     ping_timer: tokio::time::Interval,
@@ -97,7 +104,12 @@ impl FirehoseStream {
             .parse()
             .map_err(|e| FirehoseError::InvalidUri(format!("{e}")))?;
 
-        let builder = ClientBuilder::from_uri(uri);
+        let mut builder = ClientBuilder::from_uri(uri);
+        if scheme == "wss"
+            && let Some(connector) = TLS_CONNECTOR.as_ref()
+        {
+            builder = builder.connector(connector);
+        }
         let (ws, _) = if public {
             builder.resolver(PublicResolver::system()).connect().await?
         } else {
-- 
2.51.2