Auto merge of #1864 - jtgeibel:update/conduit-hyper, r=carols10cents

Switch to async/await for bin/server and tests

This PR series converts the future combinations to async/await syntax in the server binary (when `USE_HYPER=1` is set) and the recording HTTP proxy used in tests.

This change requires the update of several dependencies at once (see the list below).  It is not strictly necessary to update `reqwest` at this time, but I'm doing so in this series to avoid the build time regression of pulling in `hyper 0.12`, `hyper 0.13` and their associated transitive dependencies.

Blockers:

- [x] async/await is stable in 1.39.0 🎉
- [x] Bump the rust release in `RustConfig`
- Stable release of underlying stack
  - [x] `futures 0.3.1`
  - [x] `tokio 0.2.9`
  - [x] `hyper 0.13.1` and `hyper-tls 0.4.1`
  - [x] `reqwest 0.10.1`
  - [x] `conduit-hyper 0.2.0`

Cargo.toml

@@ -57,7 +57,7 @@ ammonia = "3.0.0"
 docopt = "1.0"
 scheduled-thread-pool = "0.2.0"
 derive_deref = "1.0.0"
-reqwest = "0.9.1"
+reqwest = { version = "0.10", features = ["blocking", "gzip", "json"] }
 tempfile = "3"
 parking_lot = "0.7.1"
 jemallocator = { version = "0.3", features = ['unprefixed_malloc_on_supported_platforms', 'profiling'] }
@@ -75,21 +75,22 @@ conduit-router = "0.8"
 conduit-static = "0.8"
 conduit-git-http-backend = "0.8"
 civet = "0.9"
-conduit-hyper = "0.1.3"
+conduit-hyper = "0.2.0"
 
-futures = "0.1"
-tokio = "0.1"
-hyper = "0.12"
+futures = "0.3"
+tokio = { version = "0.2", default-features = false, features = ["net", "signal", "io-std"]}
+hyper = "0.13"
 ctrlc = { version = "3.0", features = ["termination"] }
 indexmap = "1.0.2"
 handlebars = "2.0.1"
 
 [dev-dependencies]
 conduit-test = "0.8"
-hyper-tls = "0.3"
+hyper-tls = "0.4"
 lazy_static = "1.0"
-tokio-core = "0.1"
 diesel_migrations = { version = "1.3.0", features = ["postgres"] }
+tower-service = "0.3.0"
+tokio = { version = "0.2", default-features = false, features = ["stream"]}
 
 [build-dependencies]
 dotenv = "0.15"

src/app.rs

@@ -5,7 +5,7 @@ use std::{path::PathBuf, sync::Arc, time::Duration};
 
 use diesel::r2d2;
 use oauth2::basic::BasicClient;
-use reqwest::Client;
+use reqwest::blocking::Client;
 use scheduled_thread_pool::ScheduledThreadPool;
 
 /// The `App` struct holds the main components of the application like

src/background_jobs.rs

@@ -1,3 +1,4 @@
+use reqwest::blocking::Client;
 use std::panic::AssertUnwindSafe;
 use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
 
@@ -26,7 +27,7 @@ pub struct Environment {
     // FIXME: https://github.com/sfackler/r2d2/pull/70
     pub connection_pool: AssertUnwindSafe<DieselPool>,
     pub uploader: Uploader,
-    http_client: AssertUnwindSafe<reqwest::Client>,
+    http_client: AssertUnwindSafe<Client>,
 }
 
 // FIXME: AssertUnwindSafe should be `Clone`, this can be replaced with
@@ -47,7 +48,7 @@ impl Environment {
         index: Repository,
         connection_pool: DieselPool,
         uploader: Uploader,
-        http_client: reqwest::Client,
+        http_client: Client,
     ) -> Self {
         Self {
             index: Arc::new(Mutex::new(index)),
@@ -68,7 +69,7 @@ impl Environment {
     }
 
     /// Returns a client for making HTTP requests to upload crate files.
-    pub(crate) fn http_client(&self) -> &reqwest::Client {
+    pub(crate) fn http_client(&self) -> &Client {
         &self.http_client
     }
 }

src/bin/background-worker.rs

@@ -15,6 +15,7 @@
 use cargo_registry::git::{Repository, RepositoryConfig};
 use cargo_registry::{background_jobs::*, db};
 use diesel::r2d2;
+use reqwest::blocking::Client;
 use std::thread::sleep;
 use std::time::Duration;
 
@@ -43,12 +44,7 @@ fn main() {
     let repository = Repository::open(&repository_config).expect("Failed to clone index");
     println!("Index cloned");
 
-    let environment = Environment::new(
-        repository,
-        db_pool.clone(),
-        config.uploader,
-        reqwest::Client::new(),
-    );
+    let environment = Environment::new(repository, db_pool.clone(), config.uploader, Client::new());
 
     let build_runner = || {
         swirl::Runner::builder(db_pool.clone(), environment.clone())

src/bin/on_call/mod.rs

@@ -1,6 +1,6 @@
 use cargo_registry::util::Error;
 
-use reqwest::{header, StatusCode as Status};
+use reqwest::{blocking::Client, header, StatusCode as Status};
 
 #[derive(serde::Serialize, Debug)]
 #[serde(rename_all = "snake_case", tag = "event_type")]
@@ -29,7 +29,7 @@ impl Event {
         let api_token = dotenv::var("PAGERDUTY_API_TOKEN")?;
         let service_key = dotenv::var("PAGERDUTY_INTEGRATION_KEY")?;
 
-        let mut response = reqwest::Client::new()
+        let response = Client::new()
             .post("https://events.pagerduty.com/generic/2010-04-15/create_event.json")
             .header(header::ACCEPT, "application/vnd.pagerduty+json;version=2")
             .header(header::AUTHORIZATION, format!("Token token={}", api_token))

src/bin/render-readmes.rs

@@ -21,7 +21,7 @@ use chrono::{TimeZone, Utc};
 use diesel::{dsl::any, prelude::*};
 use docopt::Docopt;
 use flate2::read::GzDecoder;
-use reqwest::{header, Client};
+use reqwest::{blocking::Client, header};
 use tar::{self, Archive};
 
 const CACHE_CONTROL_README: &str = "public,max-age=604800";
@@ -170,7 +170,7 @@ fn get_readme(
         .uploader
         .crate_location(krate_name, &version.num.to_string());
 
-    let mut response = match client.get(&location).send() {
+    let response = match client.get(&location).send() {
         Ok(r) => r,
         Err(err) => {
             println!(

src/bin/server.rs

@@ -9,18 +9,18 @@ use std::{
 };
 
 use civet::Server as CivetServer;
-use conduit_hyper::Service as HyperService;
-use futures::Future;
-use reqwest::Client;
+use conduit_hyper::Service;
+use futures::prelude::*;
+use reqwest::blocking::Client;
 
 enum Server {
     Civet(CivetServer),
-    Hyper(tokio::runtime::Runtime),
+    Hyper(tokio::runtime::Runtime, tokio::task::JoinHandle<()>),
 }
 
 use Server::*;
 
-fn main() {
+fn main() -> Result<(), Box<dyn std::error::Error>> {
     // Initialize logging
     env_logger::init();
 
@@ -57,35 +57,43 @@ fn main() {
         });
 
     let server = if dotenv::var("USE_HYPER").is_ok() {
-        println!("Booting with a hyper based server");
-        let addr = ([127, 0, 0, 1], port).into();
-        let service = HyperService::new(app, threads as usize);
-        let server = hyper::Server::bind(&addr).serve(service);
-
-        let (tx, rx) = futures::sync::oneshot::channel::<()>();
-        let server = server
-            .with_graceful_shutdown(rx)
-            .map_err(|e| log::error!("Server error: {}", e));
+        use tokio::io::AsyncWriteExt;
+        use tokio::signal::unix::{signal, SignalKind};
 
-        ctrlc_handler(move || tx.send(()).unwrap_or(()));
+        println!("Booting with a hyper based server");
 
         let mut rt = tokio::runtime::Builder::new()
-            .core_threads(4)
-            .name_prefix("hyper-server-worker-")
-            .after_start(|| {
-                log::debug!("Stared thread {}", thread::current().name().unwrap_or("?"))
-            })
-            .before_stop(|| {
-                log::debug!(
-                    "Stopping thread {}",
-                    thread::current().name().unwrap_or("?")
-                )
-            })
+            .threaded_scheduler()
+            .enable_all()
             .build()
             .unwrap();
-        rt.spawn(server);
 
-        Hyper(rt)
+        let handler = Arc::new(conduit_hyper::BlockingHandler::new(app, threads as usize));
+        let make_service =
+            hyper::service::make_service_fn(move |socket: &hyper::server::conn::AddrStream| {
+                let addr = socket.remote_addr();
+                let handler = handler.clone();
+                async move { Service::from_blocking(handler, addr) }
+            });
+
+        let addr = ([127, 0, 0, 1], port).into();
+        let server = rt.block_on(async { hyper::Server::bind(&addr).serve(make_service) });
+
+        let mut sig_int = rt.block_on(async { signal(SignalKind::interrupt()) })?;
+        let mut sig_term = rt.block_on(async { signal(SignalKind::terminate()) })?;
+
+        let server = server.with_graceful_shutdown(async move {
+            // Wait for either signal
+            futures::select! {
+                _ = sig_int.recv().fuse() => (),
+                _ = sig_term.recv().fuse() => (),
+            };
+            let mut stdout = tokio::io::stdout();
+            stdout.write_all(b"Starting graceful shutdown\n").await.ok();
+        });
+
+        let server = rt.spawn(async { server.await.unwrap() });
+        Hyper(rt, server)
     } else {
         println!("Booting with a civet based server");
         let mut cfg = civet::Config::new();
@@ -112,7 +120,9 @@ fn main() {
 
     // Block the main thread until the server has shutdown
     match server {
-        Hyper(rt) => rt.shutdown_on_idle().wait().unwrap(),
+        Hyper(mut rt, server) => {
+            rt.block_on(async { server.await.unwrap() });
+        }
         Civet(server) => {
             let (tx, rx) = channel::<()>();
             ctrlc_handler(move || tx.send(()).unwrap_or(()));
@@ -122,6 +132,7 @@ fn main() {
     }
 
     println!("Server has gracefully shutdown!");
+    Ok(())
 }
 
 fn ctrlc_handler<F>(f: F)

src/github.rs

@@ -24,6 +24,7 @@ where
         .get(&url)
         .header(header::ACCEPT, "application/vnd.github.v3+json")
         .header(header::AUTHORIZATION, format!("token {}", auth.secret()))
+        .header(header::USER_AGENT, "crates.io (https://crates.io)")
         .send()?
         .error_for_status()
         .map_err(|e| handle_error_response(&e))?

src/s3/Cargo.toml

@@ -17,4 +17,4 @@ path = "lib.rs"
 base64 = "0.6"
 chrono = "0.4"
 openssl = "0.10.13"
-reqwest = "0.9.1"
+reqwest = { version = "0.10", features = ["blocking"] }

src/s3/lib.rs

@@ -6,7 +6,10 @@ use openssl::error::ErrorStack;
 use openssl::hash::MessageDigest;
 use openssl::pkey::PKey;
 use openssl::sign::Signer;
-use reqwest::{header, Body, Client, Response};
+use reqwest::{
+    blocking::{Body, Client, Response},
+    header,
+};
 
 mod error;
 pub use error::Error;
@@ -60,6 +63,7 @@ impl Bucket {
             .header(header::AUTHORIZATION, auth)
             .header(header::CONTENT_TYPE, content_type)
             .header(header::DATE, date)
+            .header(header::USER_AGENT, "crates.io (https://crates.io)")
             .headers(extra_headers)
             .body(Body::sized(content, content_length))
             .send()?

src/tasks/dump_db.rs

@@ -152,7 +152,7 @@ impl DumpTarball {
     }
 
     fn upload(&self, target_name: &str, uploader: &Uploader) -> Result<u64, PerformError> {
-        let client = reqwest::Client::new();
+        let client = reqwest::blocking::Client::new();
         let tarfile = File::open(&self.tarball_path)?;
         let content_length = tarfile.metadata()?.len();
         // TODO Figure out the correct content type.

src/tests/all.rs

@@ -29,7 +29,7 @@ use std::{
 
 use conduit_test::MockRequest;
 use diesel::prelude::*;
-use reqwest::{Client, Proxy};
+use reqwest::{blocking::Client, Proxy};
 
 macro_rules! t {
     ($e:expr) => {

src/tests/http-data/krate_good_badges

@@ -20,10 +20,6 @@
           "authorization",
           "AWS AKIAICL5IWUZYWWKA7JA:kDm23yhf8YuOKpTcbHhNBa6BtQw="
         ],
-        [
-          "user-agent",
-          "reqwest/0.9.1"
-        ],
         [
           "host",
           "alexcrichton-test.s3.amazonaws.com"

src/tests/http-data/krate_good_categories

@@ -28,10 +28,6 @@
           "date",
           "Fri, 15 Sep 2017 07:53:05 -0700"
         ],
-        [
-          "user-agent",
-          "reqwest/0.9.1"
-        ],
         [
           "accept",
           "*/*"

src/tests/http-data/krate_ignored_badges

@@ -20,10 +20,6 @@
           "content-length",
           "35"
         ],
-        [
-          "user-agent",
-          "reqwest/0.9.1"
-        ],
         [
           "accept-encoding",
           "gzip"

src/tests/http-data/krate_ignored_categories

@@ -8,10 +8,6 @@
           "content-length",
           "35"
         ],
-        [
-          "user-agent",
-          "reqwest/0.9.1"
-        ],
         [
           "authorization",
           "AWS AKIAICL5IWUZYWWKA7JA:V37kbEzeh57sB4yTSZIOJACPoP4="

src/tests/http-data/krate_new_krate

@@ -20,10 +20,6 @@
           "accept-encoding",
           "gzip"
         ],
-        [
-          "user-agent",
-          "reqwest/0.9.1"
-        ],
         [
           "content-type",
           "application/x-tar"

src/tests/http-data/krate_new_krate_git_upload

@@ -31,10 +31,6 @@
         [
           "accept",
           "*/*"
-        ],
-        [
-          "user-agent",
-          "reqwest/0.9.1"
         ]
       ],
       "body": "H4sIAAAAAAAA/+3AAQEAAACCIP+vbkhQwKsBLq+17wAEAAA="

src/tests/http-data/krate_new_krate_git_upload_appends

@@ -20,10 +20,6 @@
           "accept",
           "*/*"
         ],
-        [
-          "user-agent",
-          "reqwest/0.9.1"
-        ],
         [
           "authorization",
           "AWS AKIAICL5IWUZYWWKA7JA:UgUqqHJ9cQAZDdbcsxpnC0BI2eE="
@@ -91,10 +87,6 @@
           "accept",
           "*/*"
         ],
-        [
-          "user-agent",
-          "reqwest/0.9.1"
-        ],
         [
           "authorization",
           "AWS AKIAICL5IWUZYWWKA7JA:UgUqqHJ9cQAZDdbcsxpnC0BI2eE="

src/tests/http-data/krate_new_krate_git_upload_with_conflicts

@@ -24,10 +24,6 @@
           "authorization",
           "AWS AKIAICL5IWUZYWWKA7JA:241ftMxnamoj94RBOB/al86Xwjk="
         ],
-        [
-          "user-agent",
-          "reqwest/0.9.1"
-        ],
         [
           "accept",
           "*/*"