fix(sync-service): select the downstream TLS crypto provider

Cargo.toml

@@ -447,6 +447,7 @@ utoipa = { version = "5.4.0", features = [
 ] }
 utoipa-swagger-ui = { version = "9", features = ["axum", "vendored"] }
 uuid = { version = "1.15.1", features = ["fast-rng", "serde", "v4", "v5", "v7"] }
+webpki-roots = "1.0.10"
 wiremock = "0.6.5"
 zeroize = "1.8.2"
 zune-core = "0.5"

crates/sync_router/AGENTS.md

@@ -34,6 +34,8 @@ service, Redis subscriber, or Pulumi stack.
 - Inbound messages are capped at one MiB; downstream queues share an eight MiB
   budget per session, retained through the write. Never increase count limits
   without checking the byte bound.
+- Select the Rustls provider explicitly in the downstream connector; workspace
+  features enable both providers, so automatic selection panics on `wss://`.
 - Dropping a downstream route cancels its task, including pending dials/writes.
 - Every DSS replica handles only its own sockets. No distributed route
   ownership or singleton deployment is required.

crates/sync_router/Cargo.toml

@@ -9,6 +9,7 @@ anyhow = { workspace = true }
 bebop = { workspace = true }
 futures = { workspace = true }
 opentelemetry = { workspace = true }
+rustls = { workspace = true }
 sync_service_bebop_schema = { path = "../../services/sync-service/bebop-schema", features = ["bebop-owned-all"] }
 thiserror = { workspace = true }
 tokio = { workspace = true, features = ["macros", "rt-multi-thread", "sync", "time"] }
@@ -17,6 +18,7 @@ url = { workspace = true }
 tokio-tungstenite = { workspace = true }
 tracing = { workspace = true }
 tracing-opentelemetry = { workspace = true }
+webpki-roots = { workspace = true }
 workspace-hack = { version = "0.1", path = "../workspace-hack" }
 
 [dev-dependencies]

crates/sync_router/src/outbound/do_downstream.rs

@@ -14,7 +14,10 @@ use std::sync::Arc;
 use std::time::Duration;
 use tokio::sync::{Semaphore, mpsc};
 use tokio_tungstenite::tungstenite::Message;
-use tokio_tungstenite::{connect_async_with_config, tungstenite::protocol::WebSocketConfig};
+use tokio_tungstenite::{
+    Connector, MaybeTlsStream, WebSocketStream, connect_async_tls_with_config,
+    tungstenite::protocol::WebSocketConfig,
+};
 use tokio_util::{sync::CancellationToken, task::TaskTracker};
 use tracing::{Instrument, debug, warn};
 use tracing_opentelemetry::OpenTelemetrySpanExt;
@@ -125,21 +128,8 @@ async fn pump<Sink: EdgeSink>(
         }
     };
 
-    let upstream = match tokio::time::timeout(
-        Duration::from_secs(10),
-        connect_async_with_config(
-            &url,
-            Some(
-                WebSocketConfig::default()
-                    .max_message_size(Some(crate::session::MAX_FRAME_BYTES - 1024))
-                    .max_frame_size(Some(crate::session::MAX_FRAME_BYTES - 1024)),
-            ),
-            false,
-        ),
-    )
-    .await
-    {
-        Ok(Ok((upstream, _response))) => upstream,
+    let upstream = match tokio::time::timeout(Duration::from_secs(10), connect(&url)).await {
+        Ok(Ok(upstream)) => upstream,
         failure => {
             warn!(timed_out = failure.is_err(), "downstream dial failed");
             events
@@ -218,6 +208,33 @@ async fn pump<Sink: EdgeSink>(
     }
 }
 
+/// Select the provider explicitly: workspace feature unification enables both
+/// Rustls providers, so its process-wide automatic selection would panic.
+async fn connect(
+    url: &str,
+) -> anyhow::Result<WebSocketStream<MaybeTlsStream<tokio::net::TcpStream>>> {
+    let tls = rustls::ClientConfig::builder_with_provider(
+        rustls::crypto::aws_lc_rs::default_provider().into(),
+    )
+    .with_safe_default_protocol_versions()?
+    .with_root_certificates(rustls::RootCertStore::from_iter(
+        webpki_roots::TLS_SERVER_ROOTS.iter().cloned(),
+    ))
+    .with_no_client_auth();
+    let (socket, _) = connect_async_tls_with_config(
+        url,
+        Some(
+            WebSocketConfig::default()
+                .max_message_size(Some(crate::session::MAX_FRAME_BYTES - 1024))
+                .max_frame_size(Some(crate::session::MAX_FRAME_BYTES - 1024)),
+        ),
+        false,
+        Some(Connector::Rustls(Arc::new(tls))),
+    )
+    .await?;
+    Ok(socket)
+}
+
 /// The current span's context as a W3C traceparent, if a trace is recording.
 fn current_traceparent() -> Option<String> {
     let context = tracing::Span::current().context();

crates/sync_router/src/session/test.rs

@@ -345,3 +345,38 @@ async fn all_document_queues_share_a_byte_budget_until_writes_finish() {
     second.try_send(vec![0; 10]).unwrap();
     assert_eq!(budget.available_permits(), 0);
 }
+
+#[tokio::test]
+async fn tls_handshake_failure_is_reported_without_losing_the_downstream_task() {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let base_url = Url::parse(&format!("https://{}", listener.local_addr().unwrap())).unwrap();
+    let server = tokio::spawn(async move {
+        let (mut socket, _) = listener.accept().await.unwrap();
+        // Receive a TLS handshake record, then reject the connection.
+        assert_eq!(socket.read_u8().await.unwrap(), 0x16);
+    });
+    let (sender, mut receiver) = mpsc::channel(10);
+    let session = SyncSessionFactory::new(base_url)
+        .unwrap()
+        .open("socket".into(), sender);
+    session
+        .frame(encode(ToRouter::RouterSubscribe {
+            doc_id: "document",
+            token: "token",
+        }))
+        .await
+        .unwrap();
+    let frame = tokio::time::timeout(std::time::Duration::from_secs(2), receiver.recv())
+        .await
+        .unwrap()
+        .unwrap();
+    assert!(matches!(
+        FromRouter::deserialize(&frame).unwrap(),
+        FromRouter::RouterSubscribeFailed {
+            doc_id: "document",
+            ..
+        }
+    ));
+    session.shutdown().await;
+    server.await.unwrap();
+}