Merge pull request #138 from zhaofengli/axum-0.7

Various bumps

.editorconfig

@@ -13,7 +13,7 @@ charset = utf-8
 # Rust
 [*.rs]
 indent_style = space
-indent_size = 2
+indent_size = 4
 
 # Misc
 [*.{yaml,yml,md,nix}]

attic/Cargo.toml

@@ -6,7 +6,7 @@ publish = false
 
 [dependencies]
 async-stream = { version = "0.3.5", optional = true }
-base64 = "0.21.2"
+base64 = "0.22.1"
 bytes = "1.4.0"
 displaydoc = "0.2.4"
 digest = "0.10.7"
@@ -53,7 +53,7 @@ default = [ "nix_store", "tokio" ]
 # Native libnixstore bindings.
 #
 # When disabled, the native Rust portions of nix_store can still be used.
-nix_store = [ "dep:cxx", "dep:cxx-build" ]
+nix_store = [ "dep:cxx", "dep:cxx-build", "tokio/rt" ]
 
 # Tokio.
 #

attic/build.rs

@@ -24,7 +24,7 @@ fn build_bridge() {
         .flag("nix/config.h")
         .flag("-idirafter")
         .flag(hacky_include.path().to_str().unwrap())
-        // In Nix 2.19+, nix/args/root.hh depends on being able to #include "root.hh" (which is in its parent directory), for some reason
+        // In Nix 2.19+, nix/args/root.hh depends on being able to #include "args.hh" (which is in its parent directory), for some reason
         .flag("-I")
         .flag(concat!(env!("NIX_INCLUDE_PATH"), "/nix"))
         .compile("nixbinding");

client/Cargo.toml

@@ -12,7 +12,7 @@ path = "src/main.rs"
 attic = { path = "../attic" }
 
 anyhow = "1.0.71"
-async-channel = "1.8.0"
+async-channel = "2.3.1"
 bytes = "1.4.0"
 clap = { version = "4.3", features = ["derive"] }
 clap_complete = "4.3.0"
@@ -26,7 +26,7 @@ indicatif = "0.17.3"
 lazy_static = "1.4.0"
 notify = { version = "6.0.0", default-features = false, features = ["macos_kqueue"] }
 regex = "1.8.3"
-reqwest = { version = "0.11.18", default-features = false, features = ["json", "rustls-tls", "rustls-tls-native-roots", "stream"] }
+reqwest = { version = "0.12.4", default-features = false, features = ["json", "rustls-tls", "rustls-tls-native-roots", "stream"] }
 serde = { version = "1.0.163", features = ["derive"] }
 serde_json = "1.0.96"
 toml = "0.8.8"

flake.lock

@@ -7,11 +7,11 @@
         ]
       },
       "locked": {
-        "lastModified": 1702918879,
+        "lastModified": 1717025063,
-        "narHash": "sha256-tWJqzajIvYcaRWxn+cLUB9L9Pv4dQ3Bfit/YjU5ze3g=",
+        "narHash": "sha256-dIubLa56W9sNNz0e8jGxrX3CAkPXsq7snuFA/Ie6dn8=",
         "owner": "ipetkov",
         "repo": "crane",
-        "rev": "7195c00c272fdd92fc74e7d5a0a2844b9fadb2fb",
+        "rev": "480dff0be03dac0e51a8dfc26e882b0d123a450e",
         "type": "github"
       },
       "original": {

server/Cargo.toml

@@ -25,11 +25,11 @@ attic-token = { path = "../token" }
 anyhow = "1.0.71"
 async-stream = "0.3.5"
 async-trait = "0.1.68"
-aws-config = "0.57.1"
+aws-config = "1.5.0"
-aws-sdk-s3 = "0.35.0"
+aws-sdk-s3 = "1.32.0"
-axum = "0.6.18"
+axum = "0.7.5"
-axum-macros = "0.3.7"
+axum-macros = "0.4.1"
-base64 = "0.21.2"
+base64 = "0.22.1"
 bytes = "1.4.0"
 chrono = "0.4.24"
 clap = { version = "4.3", features = ["derive"] }
@@ -40,6 +40,7 @@ enum-as-inner = "0.6.0"
 fastcdc = "3.0.3"
 futures = "0.3.28"
 hex = "0.4.3"
+http-body-util = "0.1.1"
 humantime = "2.1.0"
 humantime-serde = "1.1.1"
 itoa = "=1.0.5"
@@ -53,7 +54,7 @@ serde_json = "1.0.96"
 serde_with = "3.0.0"
 tokio-util = { version = "0.7.8", features = [ "io" ] }
 toml = "0.8.8"
-tower-http = { version = "0.4.0", features = [ "catch-panic", "trace" ] }
+tower-http = { version = "0.5.2", features = [ "catch-panic", "trace" ] }
 tracing = "0.1.37"
 tracing-error = "0.2.0"
 tracing-subscriber = { version = "0.3.17", features = [ "json" ] }

server/src/access/http.rs

@@ -1,6 +1,6 @@
 //! HTTP middlewares for access control.
 
-use axum::{http::Request, middleware::Next, response::Response};
+use axum::{extract::Request, middleware::Next, response::Response};
 use sea_orm::DatabaseConnection;
 use tokio::sync::OnceCell;
 
@@ -93,7 +93,7 @@ impl AuthState {
 }
 
 /// Performs auth.
-pub async fn apply_auth<B>(req: Request<B>, next: Next<B>) -> Response {
+pub async fn apply_auth(req: Request, next: Next) -> Response {
     let token: Option<Token> = req
         .headers()
         .get("Authorization")

server/src/api/binary_cache.rs

@@ -9,9 +9,8 @@ use std::io::{Error as IoError, ErrorKind as IoErrorKind};
 use std::path::PathBuf;
 use std::sync::Arc;
 
-use axum::body::HttpBody as _;
 use axum::{
-    body::StreamBody,
+    body::Body,
     extract::{Extension, Path},
     http::StatusCode,
     response::{IntoResponse, Redirect, Response},
@@ -19,6 +18,7 @@ use axum::{
     Router,
 };
 use futures::stream::BoxStream;
+use http_body_util::BodyExt;
 use serde::Serialize;
 use tokio_util::io::ReaderStream;
 use tracing::instrument;
@@ -218,10 +218,10 @@ async fn get_nar(
             Download::Url(url) => Ok(Redirect::temporary(&url).into_response()),
             Download::AsyncRead(stream) => {
                 let stream = ReaderStream::new(stream);
-                let body = StreamBody::new(stream).map_err(|e| {
+                let body = Body::from_stream(stream).map_err(|e| {
                     tracing::error!("Stream error: {e}");
                     e
-                });
+                }).into_inner();
 
                 Ok(body.into_response())
             }
@@ -255,10 +255,10 @@ async fn get_nar(
         // TODO: Make num_prefetch configurable
         // The ideal size depends on the average chunk size
         let merged = merge_chunks(chunks, streamer, storage, 2);
-        let body = StreamBody::new(merged).map_err(|e| {
+        let body = Body::from_stream(merged).map_err(|e| {
             tracing::error!("Stream error: {e}");
             e
-        });
+        }).into_inner();
 
         Ok(body.into_response())
     }

server/src/api/v1/upload_path.rs

@@ -8,7 +8,8 @@ use anyhow::anyhow;
 use async_compression::tokio::bufread::{BrotliEncoder, XzEncoder, ZstdEncoder};
 use async_compression::Level as CompressionLevel;
 use axum::{
-    extract::{BodyStream, Extension, Json},
+    body::Body,
+    extract::{Extension, Json},
     http::HeaderMap,
 };
 use bytes::{Bytes, BytesMut};
@@ -120,8 +121,9 @@ pub(crate) async fn upload_path(
     Extension(state): Extension<State>,
     Extension(req_state): Extension<RequestState>,
     headers: HeaderMap,
-    stream: BodyStream,
+    body: Body,
 ) -> ServerResult<Json<UploadPathResult>> {
+    let stream = body.into_data_stream();
     let mut stream = StreamReader::new(
         stream.map(|r| r.map_err(|e| io::Error::new(io::ErrorKind::Other, e.to_string()))),
     );

server/src/lib.rs

@@ -26,6 +26,7 @@ pub mod nix_manifest;
 pub mod oobe;
 mod storage;
 
+use std::future::IntoFuture;
 use std::net::SocketAddr;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
@@ -38,6 +39,7 @@ use axum::{
     Router,
 };
 use sea_orm::{query::Statement, ConnectionTrait, Database, DatabaseConnection};
+use tokio::net::TcpListener;
 use tokio::sync::OnceCell;
 use tokio::time;
 use tower_http::catch_panic::CatchPanicLayer;
@@ -221,8 +223,10 @@ pub async fn run_api_server(cli_listen: Option<SocketAddr>, config: Config) -> R
 
     eprintln!("Listening on {:?}...", listen);
 
+    let listener = TcpListener::bind(&listen).await?;
+
     let (server_ret, _) = tokio::join!(
-        axum::Server::bind(&listen).serve(rest.into_make_service()),
+        axum::serve(listener, rest).into_future(),
         async {
             if state.config.database.heartbeat {
                 let _ = state.run_db_heartbeat().await;

server/src/middleware.rs

@@ -3,8 +3,8 @@ use std::sync::Arc;
 
 use anyhow::anyhow;
 use axum::{
-    extract::{Extension, Host},
+    extract::{Extension, Host, Request},
-    http::{HeaderValue, Request},
+    http::HeaderValue,
     middleware::Next,
     response::Response,
 };
@@ -14,11 +14,11 @@ use crate::error::{ErrorKind, ServerResult};
 use attic::api::binary_cache::ATTIC_CACHE_VISIBILITY;
 
 /// Initializes per-request state.
-pub async fn init_request_state<B>(
+pub async fn init_request_state(
     Extension(state): Extension<State>,
     Host(host): Host,
-    mut req: Request<B>,
+    mut req: Request,
-    next: Next<B>,
+    next: Next,
 ) -> Response {
     // X-Forwarded-Proto is an untrusted header
     let client_claims_https =
@@ -45,11 +45,11 @@ pub async fn init_request_state<B>(
 ///
 /// We also require that all request have a Host header in
 /// the first place.
-pub async fn restrict_host<B>(
+pub async fn restrict_host(
     Extension(state): Extension<State>,
     Host(host): Host,
-    req: Request<B>,
+    req: Request,
-    next: Next<B>,
+    next: Next,
 ) -> ServerResult<Response> {
     let allowed_hosts = &state.config.allowed_hosts;
 
@@ -61,10 +61,10 @@ pub async fn restrict_host<B>(
 }
 
 /// Sets the `X-Attic-Cache-Visibility` header in responses.
-pub(crate) async fn set_visibility_header<B>(
+pub(crate) async fn set_visibility_header(
     Extension(req_state): Extension<RequestState>,
-    req: Request<B>,
+    req: Request,
-    next: Next<B>,
+    next: Next,
 ) -> ServerResult<Response> {
     let mut response = next.run(req).await;
 

server/src/storage/s3.rs

@@ -3,6 +3,7 @@
 use std::time::Duration;
 
 use async_trait::async_trait;
+use aws_config::BehaviorVersion;
 use aws_sdk_s3::{
     config::Builder as S3ConfigBuilder,
     config::{Credentials, Region},
@@ -91,7 +92,7 @@ impl S3Backend {
     }
 
     async fn config_builder(config: &S3StorageConfig) -> ServerResult<S3ConfigBuilder> {
-        let shared_config = aws_config::load_from_env().await;
+        let shared_config = aws_config::load_defaults(BehaviorVersion::v2024_03_28()).await;
         let mut builder = S3ConfigBuilder::from(&shared_config);
 
         if let Some(credentials) = &config.credentials {

token/Cargo.toml

@@ -8,7 +8,7 @@ edition = "2021"
 [dependencies]
 attic = { path = "../attic", default-features = false }
 
-base64 = "0.21.2"
+base64 = "0.22.1"
 chrono = "0.4.24"
 displaydoc = "0.2.4"
 indexmap = { version = "2.2.6", features = ["serde"] }