fix(tls): flush send buffer in the background after closing TLS stream (denoland#10146)

In denoland#9118, TLS streams were split into a "read half" and a "write half"
using tokio::io::split() to allow concurrent Conn#read() and
Conn#write() calls without one blocking the other. However, this
introduced a bug: outgoing data gets discarded when the TLS stream is
gracefully closed, because the read half is closed too early, before all
TLS control data has been received.

Fixes: denoland#9692
Fixes: denoland#10049
Fixes: denoland#10296
Fixes: denoland/std#750

Author: piscisaureus

Cargo.lock

@@ -5,7 +5,9 @@ use deno_core::serde_json;
 use deno_core::url;
 use deno_runtime::deno_fetch::reqwest;
 use deno_runtime::deno_websocket::tokio_tungstenite;
-use rustls::Session;
+use deno_runtime::ops::tls::rustls;
+use deno_runtime::ops::tls::webpki;
+use deno_runtime::ops::tls::TlsStream;
 use std::fs;
 use std::io::BufReader;
 use std::io::Cursor;
@@ -14,8 +16,7 @@ use std::process::Command;
 use std::sync::Arc;
 use tempfile::TempDir;
 use test_util as util;
-use tokio_rustls::rustls;
-use tokio_rustls::webpki;
+use tokio::task::LocalSet;
 
 #[test]
 fn js_unit_tests_lint() {
@@ -6134,79 +6135,103 @@ console.log("finish");
 
 #[tokio::test]
 async fn listen_tls_alpn() {
-  let child = util::deno_cmd()
-    .current_dir(util::root_path())
-    .arg("run")
-    .arg("--unstable")
-    .arg("--quiet")
-    .arg("--allow-net")
-    .arg("--allow-read")
-    .arg("./cli/tests/listen_tls_alpn.ts")
-    .arg("4504")
-    .stdout(std::process::Stdio::piped())
-    .spawn()
-    .unwrap();
-  let mut stdout = child.stdout.unwrap();
-  let mut buffer = [0; 5];
-  let read = stdout.read(&mut buffer).unwrap();
-  assert_eq!(read, 5);
-  let msg = std::str::from_utf8(&buffer).unwrap();
-  assert_eq!(msg, "READY");
-
-  let mut cfg = rustls::ClientConfig::new();
-  let reader =
-    &mut BufReader::new(Cursor::new(include_bytes!("./tls/RootCA.crt")));
-  cfg.root_store.add_pem_file(reader).unwrap();
-  cfg.alpn_protocols.push("foobar".as_bytes().to_vec());
-
-  let tls_connector = tokio_rustls::TlsConnector::from(Arc::new(cfg));
-  let hostname = webpki::DNSNameRef::try_from_ascii_str("localhost").unwrap();
-  let stream = tokio::net::TcpStream::connect("localhost:4504")
-    .await
-    .unwrap();
+  // TLS streams require the presence of an ambient local task set to gracefully
+  // close dropped connections in the background.
+  LocalSet::new()
+    .run_until(async {
+      let mut child = util::deno_cmd()
+        .current_dir(util::root_path())
+        .arg("run")
+        .arg("--unstable")
+        .arg("--quiet")
+        .arg("--allow-net")
+        .arg("--allow-read")
+        .arg("./cli/tests/listen_tls_alpn.ts")
+        .arg("4504")
+        .stdout(std::process::Stdio::piped())
+        .spawn()
+        .unwrap();
+      let stdout = child.stdout.as_mut().unwrap();
+      let mut buffer = [0; 5];
+      let read = stdout.read(&mut buffer).unwrap();
+      assert_eq!(read, 5);
+      let msg = std::str::from_utf8(&buffer).unwrap();
+      assert_eq!(msg, "READY");
+
+      let mut cfg = rustls::ClientConfig::new();
+      let reader =
+        &mut BufReader::new(Cursor::new(include_bytes!("./tls/RootCA.crt")));
+      cfg.root_store.add_pem_file(reader).unwrap();
+      cfg.alpn_protocols.push("foobar".as_bytes().to_vec());
+      let cfg = Arc::new(cfg);
+
+      let hostname =
+        webpki::DNSNameRef::try_from_ascii_str("localhost").unwrap();
+
+      let tcp_stream = tokio::net::TcpStream::connect("localhost:4504")
+        .await
+        .unwrap();
+      let mut tls_stream =
+        TlsStream::new_client_side(tcp_stream, &cfg, hostname);
+      tls_stream.handshake().await.unwrap();
+      let (_, session) = tls_stream.get_ref();
 
-  let tls_stream = tls_connector.connect(hostname, stream).await.unwrap();
-  let (_, session) = tls_stream.get_ref();
+      let alpn = session.get_alpn_protocol().unwrap();
+      assert_eq!(std::str::from_utf8(alpn).unwrap(), "foobar");
 
-  let alpn = session.get_alpn_protocol().unwrap();
-  assert_eq!(std::str::from_utf8(alpn).unwrap(), "foobar");
+      child.kill().unwrap();
+      child.wait().unwrap();
+    })
+    .await;
 }
 
 #[tokio::test]
 async fn listen_tls_alpn_fail() {
-  let child = util::deno_cmd()
-    .current_dir(util::root_path())
-    .arg("run")
-    .arg("--unstable")
-    .arg("--quiet")
-    .arg("--allow-net")
-    .arg("--allow-read")
-    .arg("./cli/tests/listen_tls_alpn.ts")
-    .arg("4505")
-    .stdout(std::process::Stdio::piped())
-    .spawn()
-    .unwrap();
-  let mut stdout = child.stdout.unwrap();
-  let mut buffer = [0; 5];
-  let read = stdout.read(&mut buffer).unwrap();
-  assert_eq!(read, 5);
-  let msg = std::str::from_utf8(&buffer).unwrap();
-  assert_eq!(msg, "READY");
-
-  let mut cfg = rustls::ClientConfig::new();
-  let reader =
-    &mut BufReader::new(Cursor::new(include_bytes!("./tls/RootCA.crt")));
-  cfg.root_store.add_pem_file(reader).unwrap();
-  cfg.alpn_protocols.push("boofar".as_bytes().to_vec());
-
-  let tls_connector = tokio_rustls::TlsConnector::from(Arc::new(cfg));
-  let hostname = webpki::DNSNameRef::try_from_ascii_str("localhost").unwrap();
-  let stream = tokio::net::TcpStream::connect("localhost:4505")
-    .await
-    .unwrap();
+  // TLS streams require the presence of an ambient local task set to gracefully
+  // close dropped connections in the background.
+  LocalSet::new()
+    .run_until(async {
+      let mut child = util::deno_cmd()
+        .current_dir(util::root_path())
+        .arg("run")
+        .arg("--unstable")
+        .arg("--quiet")
+        .arg("--allow-net")
+        .arg("--allow-read")
+        .arg("./cli/tests/listen_tls_alpn.ts")
+        .arg("4505")
+        .stdout(std::process::Stdio::piped())
+        .spawn()
+        .unwrap();
+      let stdout = child.stdout.as_mut().unwrap();
+      let mut buffer = [0; 5];
+      let read = stdout.read(&mut buffer).unwrap();
+      assert_eq!(read, 5);
+      let msg = std::str::from_utf8(&buffer).unwrap();
+      assert_eq!(msg, "READY");
+
+      let mut cfg = rustls::ClientConfig::new();
+      let reader =
+        &mut BufReader::new(Cursor::new(include_bytes!("./tls/RootCA.crt")));
+      cfg.root_store.add_pem_file(reader).unwrap();
+      cfg.alpn_protocols.push("boofar".as_bytes().to_vec());
+      let cfg = Arc::new(cfg);
+
+      let hostname =
+        webpki::DNSNameRef::try_from_ascii_str("localhost").unwrap();
+
+      let tcp_stream = tokio::net::TcpStream::connect("localhost:4505")
+        .await
+        .unwrap();
+      let mut tls_stream =
+        TlsStream::new_client_side(tcp_stream, &cfg, hostname);
+      tls_stream.handshake().await.unwrap();
+      let (_, session) = tls_stream.get_ref();
 
-  let tls_stream = tls_connector.connect(hostname, stream).await.unwrap();
-  let (_, session) = tls_stream.get_ref();
+      assert!(session.get_alpn_protocol().is_none());
 
-  assert!(session.get_alpn_protocol().is_none());
+      child.kill().unwrap();
+      child.wait().unwrap();
+    })
+    .await;
 }