Merge pull request #137 from greatest-ape/2023-04-10

Remove aquatic_http_private; update dependencies

Cargo.toml

@@ -4,7 +4,6 @@ members = [
     "aquatic_common",
     "aquatic_http",
     "aquatic_http_load_test",
-    "aquatic_http_private",
     "aquatic_http_protocol",
     "aquatic_toml_config",
     "aquatic_toml_config_derive",

TODO.md

@@ -7,11 +7,11 @@
   * thiserror?
   * CI
   * uring load test?
-
-* ws: wait for crates release of glommio with membarrier fix (PR #558)
-* Release new version
 * More non-CI integration tests?
-* Remove aquatic_http_private?
+* Non-trivial dependency updates
+  * toml v0.7
+  * syn v2.0
+  * simd-json v0.7
 
 ## Medium priority
 

aquatic_common/Cargo.toml

@@ -21,7 +21,7 @@ aquatic_toml_config.workspace = true
 ahash = "0.8"
 anyhow = "1"
 arc-swap = "1"
-duplicate = "0.4"
+duplicate = "1"
 git-testament = "0.2"
 hashbrown = "0.13"
 hex = "0.4"
@@ -37,5 +37,5 @@ toml = "0.5"
 # Optional
 glommio = { version = "0.8", optional = true }
 hwloc = { version = "0.5", optional = true }
-rustls = { version = "0.20", optional = true }
+rustls = { version = "0.21", optional = true }
 rustls-pemfile = { version = "1", optional = true }

aquatic_http/Cargo.toml

@@ -31,7 +31,7 @@ cfg-if = "1"
 either = "1"
 futures = "0.3"
 futures-lite = "1"
-futures-rustls = "0.22"
+futures-rustls = "0.24"
 glommio = "0.8"
 itoa = "1"
 libc = "0.2"
@@ -47,7 +47,7 @@ rustls-pemfile = "1"
 serde = { version = "1", features = ["derive"] }
 signal-hook = { version = "0.3" }
 slab = "0.4"
-socket2 = { version = "0.4", features = ["all"] }
+socket2 = { version = "0.5", features = ["all"] }
 
 [dev-dependencies]
 quickcheck = "1"

aquatic_http_load_test/Cargo.toml

@@ -20,14 +20,14 @@ aquatic_toml_config.workspace = true
 
 anyhow = "1"
 futures-lite = "1"
-futures-rustls = "0.22"
+futures-rustls = "0.24"
 hashbrown = "0.13"
 glommio = "0.8"
 log = "0.4"
 mimalloc = { version = "0.1", default-features = false }
 rand = { version = "0.8", features = ["small_rng"] }
 rand_distr = "0.4"
-rustls = { version = "0.20", default-features = false, features = ["logging", "dangerous_configuration"] } # TLS 1.2 disabled
+rustls = { version = "0.21", default-features = false, features = ["logging", "dangerous_configuration"] } # TLS 1.2 disabled
 serde = { version = "1", features = ["derive"] }
 
 [dev-dependencies]

aquatic_http_private/Cargo.toml

@@ -1,38 +0,0 @@
-[package]
-name = "aquatic_http_private"
-keywords = ["http", "benchmark", "peer-to-peer", "torrent", "bittorrent"]
-version.workspace = true
-authors.workspace = true
-edition.workspace = true
-license.workspace = true
-repository.workspace = true
-readme.workspace = true
-rust-version.workspace = true
-
-[lib]
-name = "aquatic_http_private"
-
-[[bin]]
-name = "aquatic_http_private"
-
-[dependencies]
-aquatic_common = { workspace = true, features = ["rustls"] }
-aquatic_http_protocol = { workspace = true, features = ["axum"] }
-aquatic_toml_config.workspace = true
-
-anyhow = "1"
-axum = { version = "0.5", default-features = false, features = ["headers", "http1", "matched-path", "original-uri"] }
-dotenv = "0.15"
-futures-util = { version = "0.3", default-features = false }
-hex = "0.4"
-hyper = "0.14"
-log = "0.4"
-mimalloc = { version = "0.1", default-features = false }
-rand = { version = "0.8", features = ["small_rng"] }
-rustls = "0.20"
-serde = { version = "1", features = ["derive"] }
-signal-hook = { version = "0.3" }
-socket2 = { version = "0.4", features = ["all"] }
-sqlx = { version = "0.6", features = [ "runtime-tokio-rustls" , "mysql" ] }
-tokio = { version = "1", features = ["full"] }
-tokio-rustls = "0.23"

aquatic_http_private/README.md

@@ -1,96 +0,0 @@
-# aquatic_http_private
-
-HTTP (over TLS) BitTorrent tracker that calls a mysql stored procedure to
-determine if requests can proceed.
-
-Work in progress.
-
-## Usage
-
-### Database setup
-
-* Create database (you will typically skip this step and use your own database):
-
-```sql
-CREATE DATABASE aquatic_db;
-```
-
-* Create aquatic user (use a better password):
-
-```sql
-CREATE USER 'aquatic'@localhost IDENTIFIED BY 'aquatic_password';
-```
-
-* Create stored procedure `aquatic_announce_v1`:
-
-```sql
--- Create stored procedure called by aquatic for each announce request.
---
--- Set output parameter p_announce_allowed determines to true to allow announce.
-CREATE OR REPLACE PROCEDURE aquatic_announce_v1 (
-    -- Canonical source ip address (IPv4/IPv6)
-    IN p_source_ip VARBINARY(16),
-    -- Source port (not port where peer says it will accept BitTorrent requests)
-    IN p_source_port SMALLINT UNSIGNED,
-    -- User agent (can be NULL)
-    IN p_user_agent TEXT,
-    -- User token extracted from announce url ('/announce/USER_TOKEN/)
-    IN p_user_token VARCHAR(255),
-    -- Hex-encoded info hash
-    IN p_info_hash CHAR(40),
-    -- Peer ID. BINARY since it can be any bytes according to spec.
-    IN p_peer_id BINARY(20),
-    -- Event (started/stopped/completed) (can be NULL)
-    IN p_event VARCHAR(9),
-    -- Bytes uploaded. Passed directly from request.
-    IN p_uploaded BIGINT UNSIGNED,
-    -- Bytes downloaded. Passed directly from request.
-    IN p_downloaded BIGINT UNSIGNED,
-    -- Bytes left
-    IN p_left BIGINT UNSIGNED,
-    -- Return true to send annonunce response. Defaults to false if not set.
-    OUT p_announce_allowed BOOLEAN,
-    -- Optional failure reason. Defaults to NULL if not set.
-    OUT p_failure_reason TEXT,
-    -- Optional warning message. Defaults to NULL if not set.
-    OUT p_warning_message TEXT
-)
-MODIFIES SQL DATA
-BEGIN
-    -- Replace with your custom code
-    SELECT true INTO p_announce_allowed;
-END
-```
-
-* Give aquatic user permission to call stored procedure:
-
-```sql
-GRANT EXECUTE ON PROCEDURE aquatic_db.aquatic_announce_v1 TO 'aquatic'@localhost;
-FLUSH PRIVILEGES;
-```
-
-`CREATE OR REPLACE PROCEDURE` command, which leaves privileges in place,
-requires MariaDB 10.1.3 or later. If your database does not support it,
-each time you want to replace the procedure, you need to drop it, then
-create it using `CREATE PROCEDURE` and grant execution privileges again.
-
-### Tracker setup
-
-* Install rust compiler and cmake
-
-* Create `.env` file with database credentials:
-
-```sh
-DATABASE_URL="mysql://aquatic:aquatic_password@localhost/aquatic_db"
-```
-
-* Build and run tracker:
-
-```sh
-# Build
-cargo build --release -p aquatic_http_private
-# Generate config file (remember to set paths to TLS cert and key)
-./target/release/aquatic_http_private -p > http-private-config.toml
-# Run tracker
-./target/release/aquatic_http_private -c http-private-config.toml
-```

aquatic_http_private/src/common.rs

@@ -1,52 +0,0 @@
-use tokio::sync::{mpsc, oneshot};
-
-use aquatic_common::CanonicalSocketAddr;
-use aquatic_http_protocol::{common::InfoHash, response::Response};
-
-use crate::{config::Config, workers::socket::db::ValidatedAnnounceRequest};
-
-#[derive(Debug)]
-pub struct ChannelAnnounceRequest {
-    pub request: ValidatedAnnounceRequest,
-    pub source_addr: CanonicalSocketAddr,
-    pub response_sender: oneshot::Sender<Response>,
-}
-
-#[derive(Clone, Copy, PartialEq, Eq, Hash, Debug)]
-pub struct RequestWorkerIndex(pub usize);
-
-impl RequestWorkerIndex {
-    pub fn from_info_hash(config: &Config, info_hash: InfoHash) -> Self {
-        Self(info_hash.0[0] as usize % config.swarm_workers)
-    }
-}
-
-pub struct ChannelRequestSender(Vec<mpsc::Sender<ChannelAnnounceRequest>>);
-
-impl ChannelRequestSender {
-    pub fn new(senders: Vec<mpsc::Sender<ChannelAnnounceRequest>>) -> Self {
-        Self(senders)
-    }
-
-    pub async fn send_to(
-        &self,
-        index: RequestWorkerIndex,
-        request: ValidatedAnnounceRequest,
-        source_addr: CanonicalSocketAddr,
-    ) -> anyhow::Result<oneshot::Receiver<Response>> {
-        let (response_sender, response_receiver) = oneshot::channel();
-
-        let request = ChannelAnnounceRequest {
-            request,
-            source_addr,
-            response_sender,
-        };
-
-        match self.0[index.0].send(request).await {
-            Ok(()) => Ok(response_receiver),
-            Err(err) => {
-                Err(anyhow::Error::new(err).context("error sending ChannelAnnounceRequest"))
-            }
-        }
-    }
-}

aquatic_http_private/src/config.rs

@@ -1,119 +0,0 @@
-use std::{net::SocketAddr, path::PathBuf};
-
-use aquatic_common::privileges::PrivilegeConfig;
-use aquatic_toml_config::TomlConfig;
-use serde::Deserialize;
-
-use aquatic_common::cli::LogLevel;
-
-/// aquatic_http_private configuration
-#[derive(Clone, Debug, PartialEq, TomlConfig, Deserialize)]
-#[serde(default, deny_unknown_fields)]
-pub struct Config {
-    /// Socket workers receive requests from the socket, parse them and send
-    /// them on to the swarm workers. They then receive responses from the
-    /// swarm workers, encode them and send them back over the socket.
-    pub socket_workers: usize,
-    /// Swarm workers receive a number of requests from socket workers,
-    /// generate responses and send them back to the socket workers.
-    pub swarm_workers: usize,
-    pub worker_channel_size: usize,
-    /// Number of database connections to establish in each socket worker
-    pub db_connections_per_worker: u32,
-    pub log_level: LogLevel,
-    pub network: NetworkConfig,
-    pub protocol: ProtocolConfig,
-    pub cleaning: CleaningConfig,
-    pub privileges: PrivilegeConfig,
-}
-
-impl Default for Config {
-    fn default() -> Self {
-        Self {
-            socket_workers: 1,
-            swarm_workers: 1,
-            worker_channel_size: 128,
-            db_connections_per_worker: 4,
-            log_level: LogLevel::default(),
-            network: NetworkConfig::default(),
-            protocol: ProtocolConfig::default(),
-            cleaning: CleaningConfig::default(),
-            privileges: PrivilegeConfig::default(),
-        }
-    }
-}
-
-impl aquatic_common::cli::Config for Config {
-    fn get_log_level(&self) -> Option<LogLevel> {
-        Some(self.log_level)
-    }
-}
-
-#[derive(Clone, Debug, PartialEq, TomlConfig, Deserialize)]
-#[serde(default, deny_unknown_fields)]
-pub struct NetworkConfig {
-    /// Bind to this address
-    pub address: SocketAddr,
-    /// Path to TLS certificate (DER-encoded X.509)
-    pub tls_certificate_path: PathBuf,
-    /// Path to TLS private key (DER-encoded ASN.1 in PKCS#8 or PKCS#1 format)
-    pub tls_private_key_path: PathBuf,
-    pub keep_alive: bool,
-}
-
-impl Default for NetworkConfig {
-    fn default() -> Self {
-        Self {
-            address: SocketAddr::from(([0, 0, 0, 0], 3000)),
-            tls_certificate_path: "".into(),
-            tls_private_key_path: "".into(),
-            keep_alive: true,
-        }
-    }
-}
-
-#[derive(Clone, Debug, PartialEq, TomlConfig, Deserialize)]
-#[serde(default, deny_unknown_fields)]
-pub struct ProtocolConfig {
-    /// Maximum number of torrents to accept in scrape request
-    pub max_scrape_torrents: usize,
-    /// Maximum number of requested peers to accept in announce request
-    pub max_peers: usize,
-    /// Ask peers to announce this often (seconds)
-    pub peer_announce_interval: usize,
-}
-
-impl Default for ProtocolConfig {
-    fn default() -> Self {
-        Self {
-            max_scrape_torrents: 100,
-            max_peers: 50,
-            peer_announce_interval: 300,
-        }
-    }
-}
-
-#[derive(Clone, Debug, PartialEq, TomlConfig, Deserialize)]
-#[serde(default, deny_unknown_fields)]
-pub struct CleaningConfig {
-    /// Clean peers this often (seconds)
-    pub torrent_cleaning_interval: u64,
-    /// Remove peers that have not announced for this long (seconds)
-    pub max_peer_age: u32,
-}
-
-impl Default for CleaningConfig {
-    fn default() -> Self {
-        Self {
-            torrent_cleaning_interval: 30,
-            max_peer_age: 360,
-        }
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::Config;
-
-    ::aquatic_toml_config::gen_serialize_deserialize_test!(Config);
-}

aquatic_http_private/src/lib.rs

@@ -1,103 +0,0 @@
-mod common;
-pub mod config;
-mod workers;
-
-use std::{collections::VecDeque, sync::Arc};
-
-use aquatic_common::{
-    privileges::PrivilegeDropper, rustls_config::create_rustls_config, PanicSentinelWatcher,
-    ServerStartInstant,
-};
-use common::ChannelRequestSender;
-use dotenv::dotenv;
-use signal_hook::{consts::SIGTERM, iterator::Signals};
-use tokio::sync::mpsc::channel;
-
-use config::Config;
-
-pub const APP_NAME: &str = "aquatic_http_private: private HTTP/TLS BitTorrent tracker";
-pub const APP_VERSION: &str = env!("CARGO_PKG_VERSION");
-
-pub fn run(config: Config) -> anyhow::Result<()> {
-    let mut signals = Signals::new([SIGTERM])?;
-
-    dotenv().ok();
-
-    let tls_config = Arc::new(create_rustls_config(
-        &config.network.tls_certificate_path,
-        &config.network.tls_private_key_path,
-    )?);
-
-    let mut request_senders = Vec::new();
-    let mut request_receivers = VecDeque::new();
-
-    for _ in 0..config.swarm_workers {
-        let (request_sender, request_receiver) = channel(config.worker_channel_size);
-
-        request_senders.push(request_sender);
-        request_receivers.push_back(request_receiver);
-    }
-
-    let (sentinel_watcher, sentinel) = PanicSentinelWatcher::create_with_sentinel();
-    let priv_dropper = PrivilegeDropper::new(config.privileges.clone(), config.socket_workers);
-
-    let server_start_instant = ServerStartInstant::new();
-
-    let mut handles = Vec::new();
-
-    for _ in 0..config.socket_workers {
-        let sentinel = sentinel.clone();
-        let config = config.clone();
-        let tls_config = tls_config.clone();
-        let request_sender = ChannelRequestSender::new(request_senders.clone());
-        let priv_dropper = priv_dropper.clone();
-
-        let handle = ::std::thread::Builder::new()
-            .name("socket".into())
-            .spawn(move || {
-                workers::socket::run_socket_worker(
-                    sentinel,
-                    config,
-                    tls_config,
-                    request_sender,
-                    priv_dropper,
-                )
-            })?;
-
-        handles.push(handle);
-    }
-
-    for _ in 0..config.swarm_workers {
-        let sentinel = sentinel.clone();
-        let config = config.clone();
-        let request_receiver = request_receivers.pop_front().unwrap();
-
-        let handle = ::std::thread::Builder::new()
-            .name("request".into())
-            .spawn(move || {
-                workers::swarm::run_swarm_worker(
-                    sentinel,
-                    config,
-                    request_receiver,
-                    server_start_instant,
-                )
-            })?;
-
-        handles.push(handle);
-    }
-
-    for signal in &mut signals {
-        match signal {
-            SIGTERM => {
-                if sentinel_watcher.panic_was_triggered() {
-                    return Err(anyhow::anyhow!("worker thread panicked"));
-                } else {
-                    return Ok(());
-                }
-            }
-            _ => unreachable!(),
-        }
-    }
-
-    Ok(())
-}

aquatic_http_private/src/main.rs

@@ -1,14 +0,0 @@
-use aquatic_common::cli::run_app_with_cli_and_config;
-use aquatic_http_private::config::Config;
-
-#[global_allocator]
-static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
-
-fn main() {
-    run_app_with_cli_and_config::<Config>(
-        aquatic_http_private::APP_NAME,
-        aquatic_http_private::APP_VERSION,
-        aquatic_http_private::run,
-        None,
-    )
-}

aquatic_http_private/src/workers/mod.rs

@@ -1,2 +0,0 @@
-pub mod socket;
-pub mod swarm;

aquatic_http_private/src/workers/socket/db.rs

@@ -1,119 +0,0 @@
-use std::net::IpAddr;
-
-use aquatic_common::CanonicalSocketAddr;
-use aquatic_http_protocol::{request::AnnounceRequest, response::FailureResponse};
-use sqlx::{Executor, MySql, Pool};
-
-#[derive(Debug)]
-pub struct ValidatedAnnounceRequest(AnnounceRequest);
-
-impl Into<AnnounceRequest> for ValidatedAnnounceRequest {
-    fn into(self) -> AnnounceRequest {
-        self.0
-    }
-}
-
-#[derive(Debug, sqlx::FromRow)]
-struct AnnounceProcedureResults {
-    announce_allowed: bool,
-    failure_reason: Option<String>,
-    warning_message: Option<String>,
-}
-
-pub async fn validate_announce_request(
-    pool: &Pool<MySql>,
-    source_addr: CanonicalSocketAddr,
-    user_agent: Option<String>,
-    user_token: String,
-    request: AnnounceRequest,
-) -> Result<(ValidatedAnnounceRequest, Option<String>), FailureResponse> {
-    match call_announce_procedure(pool, source_addr, user_agent, user_token, &request).await {
-        Ok(results) => {
-            if results.announce_allowed {
-                Ok((ValidatedAnnounceRequest(request), results.warning_message))
-            } else {
-                Err(FailureResponse::new(
-                    results
-                        .failure_reason
-                        .unwrap_or_else(|| "Not allowed".into()),
-                ))
-            }
-        }
-        Err(err) => {
-            ::log::error!("announce procedure error: {:#}", err);
-
-            Err(FailureResponse::new("Internal error"))
-        }
-    }
-}
-
-async fn call_announce_procedure(
-    pool: &Pool<MySql>,
-    source_addr: CanonicalSocketAddr,
-    user_agent: Option<String>,
-    user_token: String, // FIXME: length
-    request: &AnnounceRequest,
-) -> anyhow::Result<AnnounceProcedureResults> {
-    let mut t = pool.begin().await?;
-
-    t.execute(
-        "
-        SET
-            @p_announce_allowed = false,
-            @p_failure_reason = NULL,
-            @p_warning_message = NULL;
-        ",
-    )
-    .await?;
-
-    let q = sqlx::query(
-        "
-        CALL aquatic_announce_v1(
-            ?,
-            ?,
-            ?,
-            ?,
-            ?,
-            ?,
-            ?,
-            ?,
-            ?,
-            ?,
-            @p_announce_allowed,
-            @p_failure_reason,
-            @p_warning_message
-        );
-        ",
-    )
-    .bind(match source_addr.get().ip() {
-        IpAddr::V4(ip) => Vec::from(ip.octets()),
-        IpAddr::V6(ip) => Vec::from(ip.octets()),
-    })
-    .bind(source_addr.get().port())
-    .bind(user_agent)
-    .bind(user_token)
-    .bind(hex::encode(request.info_hash.0))
-    .bind(&request.peer_id.0[..])
-    .bind(request.event.as_str())
-    .bind(request.bytes_uploaded as u64)
-    .bind(request.bytes_downloaded as u64)
-    .bind(request.bytes_left as u64);
-
-    t.execute(q).await?;
-
-    let response = sqlx::query_as::<_, AnnounceProcedureResults>(
-        "
-        SELECT
-            @p_announce_allowed as announce_allowed,
-            @p_failure_reason as failure_reason,
-            @p_warning_message as warning_message;
-        
-        ",
-    )
-    .fetch_one(&mut t)
-    .await?;
-
-    t.commit().await?;
-
-    Ok(response)
-}

aquatic_http_private/src/workers/socket/mod.rs

@@ -1,104 +0,0 @@
-pub mod db;
-mod routes;
-mod tls;
-
-use std::{
-    net::{SocketAddr, TcpListener},
-    sync::Arc,
-};
-
-use anyhow::Context;
-use aquatic_common::{privileges::PrivilegeDropper, rustls_config::RustlsConfig, PanicSentinel};
-use axum::{extract::connect_info::Connected, routing::get, Extension, Router};
-use hyper::server::conn::AddrIncoming;
-use sqlx::mysql::MySqlPoolOptions;
-
-use self::tls::{TlsAcceptor, TlsStream};
-use crate::{common::ChannelRequestSender, config::Config};
-
-impl<'a> Connected<&'a tls::TlsStream> for SocketAddr {
-    fn connect_info(target: &'a TlsStream) -> Self {
-        target.get_remote_addr()
-    }
-}
-
-pub fn run_socket_worker(
-    _sentinel: PanicSentinel,
-    config: Config,
-    tls_config: Arc<RustlsConfig>,
-    request_sender: ChannelRequestSender,
-    priv_dropper: PrivilegeDropper,
-) -> anyhow::Result<()> {
-    let tcp_listener = create_tcp_listener(config.network.address, priv_dropper)?;
-
-    let runtime = tokio::runtime::Builder::new_current_thread()
-        .enable_all()
-        .build()?;
-
-    runtime.block_on(run_app(config, tls_config, tcp_listener, request_sender))?;
-
-    Ok(())
-}
-
-async fn run_app(
-    config: Config,
-    tls_config: Arc<RustlsConfig>,
-    tcp_listener: TcpListener,
-    request_sender: ChannelRequestSender,
-) -> anyhow::Result<()> {
-    let db_url =
-        ::std::env::var("DATABASE_URL").with_context(|| "Retrieve env var DATABASE_URL")?;
-
-    let tls_acceptor = TlsAcceptor::new(
-        tls_config,
-        AddrIncoming::from_listener(tokio::net::TcpListener::from_std(tcp_listener)?)?,
-    );
-
-    let pool = MySqlPoolOptions::new()
-        .max_connections(config.db_connections_per_worker)
-        .connect(&db_url)
-        .await?;
-
-    let app = Router::new()
-        .route("/announce/:user_token/", get(routes::announce))
-        .layer(Extension(Arc::new(config.clone())))
-        .layer(Extension(pool))
-        .layer(Extension(Arc::new(request_sender)));
-
-    axum::Server::builder(tls_acceptor)
-        .http1_keepalive(config.network.keep_alive)
-        .serve(app.into_make_service_with_connect_info::<SocketAddr>())
-        .await?;
-
-    Ok(())
-}
-
-fn create_tcp_listener(
-    addr: SocketAddr,
-    priv_dropper: PrivilegeDropper,
-) -> anyhow::Result<TcpListener> {
-    let domain = if addr.is_ipv4() {
-        socket2::Domain::IPV4
-    } else {
-        socket2::Domain::IPV6
-    };
-
-    let socket = socket2::Socket::new(domain, socket2::Type::STREAM, Some(socket2::Protocol::TCP))?;
-
-    socket
-        .set_reuse_port(true)
-        .with_context(|| "set_reuse_port")?;
-    socket
-        .set_nonblocking(true)
-        .with_context(|| "set_nonblocking")?;
-    socket
-        .bind(&addr.into())
-        .with_context(|| format!("bind to {}", addr))?;
-    socket
-        .listen(1024)
-        .with_context(|| format!("listen on {}", addr))?;
-
-    priv_dropper.after_socket_creation()?;
-
-    Ok(socket.into())
-}

aquatic_http_private/src/workers/socket/routes.rs

@@ -1,65 +0,0 @@
-use aquatic_common::CanonicalSocketAddr;
-use axum::{
-    extract::{ConnectInfo, Path, RawQuery},
-    headers::UserAgent,
-    Extension, TypedHeader,
-};
-use sqlx::mysql::MySqlPool;
-use std::{net::SocketAddr, sync::Arc};
-
-use aquatic_http_protocol::{
-    request::AnnounceRequest,
-    response::{FailureResponse, Response},
-};
-
-use crate::{
-    common::{ChannelRequestSender, RequestWorkerIndex},
-    config::Config,
-};
-
-use super::db;
-
-pub async fn announce(
-    Extension(config): Extension<Arc<Config>>,
-    Extension(pool): Extension<MySqlPool>,
-    Extension(request_sender): Extension<Arc<ChannelRequestSender>>,
-    ConnectInfo(source_addr): ConnectInfo<SocketAddr>,
-    opt_user_agent: Option<TypedHeader<UserAgent>>,
-    Path(user_token): Path<String>,
-    RawQuery(query): RawQuery,
-) -> Result<Response, FailureResponse> {
-    let query = query.ok_or_else(|| FailureResponse::new("Empty query string"))?;
-
-    let request = AnnounceRequest::from_query_string(&query)
-        .map_err(|_| FailureResponse::new("Malformed request"))?;
-
-    let swarm_worker_index = RequestWorkerIndex::from_info_hash(&config, request.info_hash);
-    let opt_user_agent = opt_user_agent.map(|header| header.as_str().to_owned());
-
-    let source_addr = CanonicalSocketAddr::new(source_addr);
-
-    let (validated_request, opt_warning_message) =
-        db::validate_announce_request(&pool, source_addr, opt_user_agent, user_token, request)
-            .await?;
-
-    let response_receiver = request_sender
-        .send_to(swarm_worker_index, validated_request, source_addr)
-        .await
-        .map_err(|err| internal_error(format!("Sending request over channel failed: {:#}", err)))?;
-
-    let mut response = response_receiver.await.map_err(|err| {
-        internal_error(format!("Receiving response over channel failed: {:#}", err))
-    })?;
-
-    if let Response::Announce(ref mut r) = response {
-        r.warning_message = opt_warning_message;
-    }
-
-    Ok(response)
-}
-
-fn internal_error(error: String) -> FailureResponse {
-    ::log::error!("{}", error);
-
-    FailureResponse::new("Internal error")
-}

aquatic_http_private/src/workers/socket/tls.rs

@@ -1,151 +0,0 @@
-//! hyper/rustls integration
-//!
-//! hyper will automatically use HTTP/2 if a client starts talking HTTP/2,
-//! otherwise HTTP/1.1 will be used.
-//!
-//! Based on https://github.com/rustls/hyper-rustls/blob/9b7b1220f74de9b249ce2b8f8b922fd00074c53b/examples/server.rs
-
-// ISC License (ISC)
-// Copyright (c) 2016, Joseph Birr-Pixton <jpixton@gmail.com>
-//
-// Permission to use, copy, modify, and/or distribute this software for
-// any purpose with or without fee is hereby granted, provided that the
-// above copyright notice and this permission notice appear in all copies.
-//
-// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL
-// WARRANTIES WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED
-// WARRANTIES OF MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE
-// AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL
-// DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR
-// PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
-// ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF
-// THIS SOFTWARE.
-
-use core::task::{Context, Poll};
-use futures_util::ready;
-use hyper::server::accept::Accept;
-use hyper::server::conn::{AddrIncoming, AddrStream};
-use std::future::Future;
-use std::io;
-use std::net::SocketAddr;
-use std::pin::Pin;
-use std::sync::Arc;
-use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
-use tokio_rustls::rustls::ServerConfig;
-
-enum State {
-    Handshaking(tokio_rustls::Accept<AddrStream>, SocketAddr),
-    Streaming(tokio_rustls::server::TlsStream<AddrStream>),
-}
-
-// tokio_rustls::server::TlsStream doesn't expose constructor methods,
-// so we have to TlsAcceptor::accept and handshake to have access to it
-// TlsStream implements AsyncRead/AsyncWrite handshaking tokio_rustls::Accept first
-pub struct TlsStream {
-    state: State,
-}
-
-impl TlsStream {
-    fn new(stream: AddrStream, config: Arc<ServerConfig>) -> TlsStream {
-        let remote_addr = stream.remote_addr();
-        let accept = tokio_rustls::TlsAcceptor::from(config).accept(stream);
-
-        TlsStream {
-            state: State::Handshaking(accept, remote_addr),
-        }
-    }
-
-    pub fn get_remote_addr(&self) -> SocketAddr {
-        match &self.state {
-            State::Handshaking(_, remote_addr) => *remote_addr,
-            State::Streaming(stream) => stream.get_ref().0.remote_addr(),
-        }
-    }
-}
-
-impl AsyncRead for TlsStream {
-    fn poll_read(
-        self: Pin<&mut Self>,
-        cx: &mut Context,
-        buf: &mut ReadBuf,
-    ) -> Poll<io::Result<()>> {
-        let pin = self.get_mut();
-        match pin.state {
-            State::Handshaking(ref mut accept, ref mut socket_addr) => {
-                match ready!(Pin::new(accept).poll(cx)) {
-                    Ok(mut stream) => {
-                        *socket_addr = stream.get_ref().0.remote_addr();
-                        let result = Pin::new(&mut stream).poll_read(cx, buf);
-                        pin.state = State::Streaming(stream);
-                        result
-                    }
-                    Err(err) => Poll::Ready(Err(err)),
-                }
-            }
-            State::Streaming(ref mut stream) => Pin::new(stream).poll_read(cx, buf),
-        }
-    }
-}
-
-impl AsyncWrite for TlsStream {
-    fn poll_write(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-        buf: &[u8],
-    ) -> Poll<io::Result<usize>> {
-        let pin = self.get_mut();
-        match pin.state {
-            State::Handshaking(ref mut accept, _) => match ready!(Pin::new(accept).poll(cx)) {
-                Ok(mut stream) => {
-                    let result = Pin::new(&mut stream).poll_write(cx, buf);
-                    pin.state = State::Streaming(stream);
-                    result
-                }
-                Err(err) => Poll::Ready(Err(err)),
-            },
-            State::Streaming(ref mut stream) => Pin::new(stream).poll_write(cx, buf),
-        }
-    }
-
-    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
-        match self.state {
-            State::Handshaking(_, _) => Poll::Ready(Ok(())),
-            State::Streaming(ref mut stream) => Pin::new(stream).poll_flush(cx),
-        }
-    }
-
-    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
-        match self.state {
-            State::Handshaking(_, _) => Poll::Ready(Ok(())),
-            State::Streaming(ref mut stream) => Pin::new(stream).poll_shutdown(cx),
-        }
-    }
-}
-
-pub struct TlsAcceptor {
-    config: Arc<ServerConfig>,
-    incoming: AddrIncoming,
-}
-
-impl TlsAcceptor {
-    pub fn new(config: Arc<ServerConfig>, incoming: AddrIncoming) -> TlsAcceptor {
-        TlsAcceptor { config, incoming }
-    }
-}
-
-impl Accept for TlsAcceptor {
-    type Conn = TlsStream;
-    type Error = io::Error;
-
-    fn poll_accept(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<Option<Result<Self::Conn, Self::Error>>> {
-        let pin = self.get_mut();
-        match ready!(Pin::new(&mut pin.incoming).poll_accept(cx)) {
-            Some(Ok(sock)) => Poll::Ready(Some(Ok(TlsStream::new(sock, pin.config.clone())))),
-            Some(Err(e)) => Poll::Ready(Some(Err(e))),
-            None => Poll::Ready(None),
-        }
-    }
-}

aquatic_http_private/src/workers/swarm/common.rs

@@ -1,121 +0,0 @@
-use std::net::{Ipv4Addr, Ipv6Addr};
-
-use aquatic_common::{IndexMap, SecondsSinceServerStart, ServerStartInstant, ValidUntil};
-use aquatic_http_protocol::common::{AnnounceEvent, InfoHash, PeerId};
-use aquatic_http_protocol::response::ResponsePeer;
-
-pub trait Ip: ::std::fmt::Debug + Copy + Eq + ::std::hash::Hash {}
-
-impl Ip for Ipv4Addr {}
-impl Ip for Ipv6Addr {}
-
-#[derive(PartialEq, Eq, Clone, Copy, Debug)]
-pub enum PeerStatus {
-    Seeding,
-    Leeching,
-    Stopped,
-}
-
-impl PeerStatus {
-    /// Determine peer status from announce event and number of bytes left.
-    ///
-    /// Likely, the last branch will be taken most of the time.
-    #[inline]
-    pub fn from_event_and_bytes_left(event: AnnounceEvent, opt_bytes_left: Option<usize>) -> Self {
-        if let AnnounceEvent::Stopped = event {
-            Self::Stopped
-        } else if let Some(0) = opt_bytes_left {
-            Self::Seeding
-        } else {
-            Self::Leeching
-        }
-    }
-}
-
-#[derive(Debug, Clone, Copy)]
-pub struct Peer<I: Ip> {
-    pub ip_address: I,
-    pub port: u16,
-    pub status: PeerStatus,
-    pub valid_until: ValidUntil,
-}
-
-impl<I: Ip> Peer<I> {
-    pub fn to_response_peer(&self) -> ResponsePeer<I> {
-        ResponsePeer {
-            ip_address: self.ip_address,
-            port: self.port,
-        }
-    }
-}
-
-#[derive(Debug, Clone, PartialEq, Eq, Hash)]
-pub struct PeerMapKey<I: Ip> {
-    pub peer_id: PeerId,
-    pub ip_address: I,
-}
-
-pub type PeerMap<I> = IndexMap<PeerMapKey<I>, Peer<I>>;
-
-pub struct TorrentData<I: Ip> {
-    pub peers: PeerMap<I>,
-    pub num_seeders: usize,
-    pub num_leechers: usize,
-}
-
-impl<I: Ip> Default for TorrentData<I> {
-    #[inline]
-    fn default() -> Self {
-        Self {
-            peers: Default::default(),
-            num_seeders: 0,
-            num_leechers: 0,
-        }
-    }
-}
-
-pub type TorrentMap<I> = IndexMap<InfoHash, TorrentData<I>>;
-
-#[derive(Default)]
-pub struct TorrentMaps {
-    pub ipv4: TorrentMap<Ipv4Addr>,
-    pub ipv6: TorrentMap<Ipv6Addr>,
-}
-
-impl TorrentMaps {
-    pub fn clean(&mut self, server_start_instant: ServerStartInstant) {
-        let now = server_start_instant.seconds_elapsed();
-
-        Self::clean_torrent_map(&mut self.ipv4, now);
-        Self::clean_torrent_map(&mut self.ipv6, now);
-    }
-
-    fn clean_torrent_map<I: Ip>(torrent_map: &mut TorrentMap<I>, now: SecondsSinceServerStart) {
-        torrent_map.retain(|_, torrent_data| {
-            let num_seeders = &mut torrent_data.num_seeders;
-            let num_leechers = &mut torrent_data.num_leechers;
-
-            torrent_data.peers.retain(|_, peer| {
-                if peer.valid_until.valid(now) {
-                    true
-                } else {
-                    match peer.status {
-                        PeerStatus::Seeding => {
-                            *num_seeders -= 1;
-                        }
-                        PeerStatus::Leeching => {
-                            *num_leechers -= 1;
-                        }
-                        _ => (),
-                    };
-
-                    false
-                }
-            });
-
-            !torrent_data.peers.is_empty()
-        });
-
-        torrent_map.shrink_to_fit();
-    }
-}

aquatic_http_private/src/workers/swarm/mod.rs

@@ -1,220 +0,0 @@
-mod common;
-
-use std::cell::RefCell;
-use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
-use std::rc::Rc;
-
-use aquatic_http_protocol::request::AnnounceRequest;
-use rand::prelude::SmallRng;
-use rand::SeedableRng;
-use tokio::sync::mpsc::Receiver;
-use tokio::task::LocalSet;
-use tokio::time;
-
-use aquatic_common::{
-    extract_response_peers, CanonicalSocketAddr, PanicSentinel, ServerStartInstant, ValidUntil,
-};
-use aquatic_http_protocol::response::{
-    AnnounceResponse, Response, ResponsePeer, ResponsePeerListV4, ResponsePeerListV6,
-};
-
-use crate::common::ChannelAnnounceRequest;
-use crate::config::Config;
-
-use common::*;
-
-pub fn run_swarm_worker(
-    _sentinel: PanicSentinel,
-    config: Config,
-    request_receiver: Receiver<ChannelAnnounceRequest>,
-    server_start_instant: ServerStartInstant,
-) -> anyhow::Result<()> {
-    let runtime = tokio::runtime::Builder::new_current_thread()
-        .enable_all()
-        .build()?;
-
-    runtime.block_on(run_inner(config, request_receiver, server_start_instant))?;
-
-    Ok(())
-}
-
-async fn run_inner(
-    config: Config,
-    mut request_receiver: Receiver<ChannelAnnounceRequest>,
-    server_start_instant: ServerStartInstant,
-) -> anyhow::Result<()> {
-    let torrents = Rc::new(RefCell::new(TorrentMaps::default()));
-    let mut rng = SmallRng::from_entropy();
-
-    LocalSet::new().spawn_local(periodically_clean_torrents(
-        config.clone(),
-        torrents.clone(),
-        server_start_instant,
-    ));
-
-    loop {
-        let request = request_receiver
-            .recv()
-            .await
-            .ok_or_else(|| anyhow::anyhow!("request channel closed"))?;
-
-        let valid_until = ValidUntil::new(server_start_instant, config.cleaning.max_peer_age);
-
-        let response = handle_announce_request(
-            &config,
-            &mut rng,
-            &mut torrents.borrow_mut(),
-            valid_until,
-            request.source_addr,
-            request.request.into(),
-        );
-
-        let _ = request.response_sender.send(Response::Announce(response));
-    }
-}
-
-async fn periodically_clean_torrents(
-    config: Config,
-    torrents: Rc<RefCell<TorrentMaps>>,
-    server_start_instant: ServerStartInstant,
-) {
-    let mut interval = time::interval(time::Duration::from_secs(
-        config.cleaning.torrent_cleaning_interval,
-    ));
-
-    loop {
-        interval.tick().await;
-
-        torrents.borrow_mut().clean(server_start_instant);
-    }
-}
-
-fn handle_announce_request(
-    config: &Config,
-    rng: &mut SmallRng,
-    torrent_maps: &mut TorrentMaps,
-    valid_until: ValidUntil,
-    source_addr: CanonicalSocketAddr,
-    request: AnnounceRequest,
-) -> AnnounceResponse {
-    match source_addr.get().ip() {
-        IpAddr::V4(source_ip) => {
-            let torrent_data: &mut TorrentData<Ipv4Addr> =
-                torrent_maps.ipv4.entry(request.info_hash).or_default();
-
-            let (seeders, leechers, response_peers) = upsert_peer_and_get_response_peers(
-                config,
-                rng,
-                torrent_data,
-                source_ip,
-                request,
-                valid_until,
-            );
-
-            let response = AnnounceResponse {
-                complete: seeders,
-                incomplete: leechers,
-                announce_interval: config.protocol.peer_announce_interval,
-                peers: ResponsePeerListV4(response_peers),
-                peers6: ResponsePeerListV6(vec![]),
-                warning_message: None,
-            };
-
-            response
-        }
-        IpAddr::V6(source_ip) => {
-            let torrent_data: &mut TorrentData<Ipv6Addr> =
-                torrent_maps.ipv6.entry(request.info_hash).or_default();
-
-            let (seeders, leechers, response_peers) = upsert_peer_and_get_response_peers(
-                config,
-                rng,
-                torrent_data,
-                source_ip,
-                request,
-                valid_until,
-            );
-
-            let response = AnnounceResponse {
-                complete: seeders,
-                incomplete: leechers,
-                announce_interval: config.protocol.peer_announce_interval,
-                peers: ResponsePeerListV4(vec![]),
-                peers6: ResponsePeerListV6(response_peers),
-                warning_message: None,
-            };
-
-            response
-        }
-    }
-}
-
-/// Insert/update peer. Return num_seeders, num_leechers and response peers
-pub fn upsert_peer_and_get_response_peers<I: Ip>(
-    config: &Config,
-    rng: &mut SmallRng,
-    torrent_data: &mut TorrentData<I>,
-    source_ip: I,
-    request: AnnounceRequest,
-    valid_until: ValidUntil,
-) -> (usize, usize, Vec<ResponsePeer<I>>) {
-    // Insert/update/remove peer who sent this request
-
-    let peer_status =
-        PeerStatus::from_event_and_bytes_left(request.event, Some(request.bytes_left));
-
-    let peer = Peer {
-        ip_address: source_ip,
-        port: request.port,
-        status: peer_status,
-        valid_until,
-    };
-
-    let peer_map_key = PeerMapKey {
-        peer_id: request.peer_id,
-        ip_address: source_ip,
-    };
-
-    let opt_removed_peer = match peer_status {
-        PeerStatus::Leeching => {
-            torrent_data.num_leechers += 1;
-
-            torrent_data.peers.insert(peer_map_key.clone(), peer)
-        }
-        PeerStatus::Seeding => {
-            torrent_data.num_seeders += 1;
-
-            torrent_data.peers.insert(peer_map_key.clone(), peer)
-        }
-        PeerStatus::Stopped => torrent_data.peers.remove(&peer_map_key),
-    };
-
-    match opt_removed_peer.map(|peer| peer.status) {
-        Some(PeerStatus::Leeching) => {
-            torrent_data.num_leechers -= 1;
-        }
-        Some(PeerStatus::Seeding) => {
-            torrent_data.num_seeders -= 1;
-        }
-        _ => {}
-    }
-
-    let max_num_peers_to_take = match request.numwant {
-        Some(0) | None => config.protocol.max_peers,
-        Some(numwant) => numwant.min(config.protocol.max_peers),
-    };
-
-    let response_peers: Vec<ResponsePeer<I>> = extract_response_peers(
-        rng,
-        &torrent_data.peers,
-        max_num_peers_to_take,
-        peer_map_key,
-        Peer::to_response_peer,
-    );
-
-    (
-        torrent_data.num_seeders,
-        torrent_data.num_leechers,
-        response_peers,
-    )
-}

aquatic_http_protocol/Cargo.toml

@@ -25,7 +25,6 @@ harness = false
 
 [dependencies]
 anyhow = "1"
-axum = { version = "0.5", optional = true, default-features = false }
 compact_str = { version = "0.7", features = ["serde"] }
 hex = { version = "0.4", default-features = false }
 httparse = "1"
@@ -38,6 +37,6 @@ urlencoding = "2"
 
 [dev-dependencies]
 bendy = { version = "0.4.0-beta.2", features = ["std", "serde"] }
-criterion = "0.3"
+criterion = "0.4"
 quickcheck = "1"
 quickcheck_macros = "1"

aquatic_http_protocol/src/response.rs

@@ -117,17 +117,6 @@ impl AnnounceResponse {
     }
 }
 
-#[cfg(feature = "axum")]
-impl axum::response::IntoResponse for AnnounceResponse {
-    fn into_response(self) -> axum::response::Response {
-        let mut response_bytes = Vec::with_capacity(128);
-
-        self.write(&mut response_bytes).unwrap();
-
-        ([("Content-type", "text/plain")], response_bytes).into_response()
-    }
-}
-
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct ScrapeResponse {
     /// BTreeMap instead of HashMap since keys need to be serialized in order
@@ -158,17 +147,6 @@ impl ScrapeResponse {
     }
 }
 
-#[cfg(feature = "axum")]
-impl axum::response::IntoResponse for ScrapeResponse {
-    fn into_response(self) -> axum::response::Response {
-        let mut response_bytes = Vec::with_capacity(128);
-
-        self.write(&mut response_bytes).unwrap();
-
-        ([("Content-type", "text/plain")], response_bytes).into_response()
-    }
-}
-
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct FailureResponse {
     #[serde(rename = "failure reason")]
@@ -197,17 +175,6 @@ impl FailureResponse {
     }
 }
 
-#[cfg(feature = "axum")]
-impl axum::response::IntoResponse for FailureResponse {
-    fn into_response(self) -> axum::response::Response {
-        let mut response_bytes = Vec::with_capacity(64);
-
-        self.write(&mut response_bytes).unwrap();
-
-        ([("Content-type", "text/plain")], response_bytes).into_response()
-    }
-}
-
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(untagged)]
 pub enum Response {
@@ -229,17 +196,6 @@ impl Response {
     }
 }
 
-#[cfg(feature = "axum")]
-impl axum::response::IntoResponse for Response {
-    fn into_response(self) -> axum::response::Response {
-        match self {
-            Self::Announce(r) => r.into_response(),
-            Self::Scrape(r) => r.into_response(),
-            Self::Failure(r) => r.into_response(),
-        }
-    }
-}
-
 #[cfg(test)]
 impl quickcheck::Arbitrary for ResponsePeer<Ipv4Addr> {
     fn arbitrary(g: &mut quickcheck::Gen) -> Self {

aquatic_udp/Cargo.toml

@@ -36,7 +36,7 @@ getrandom = "0.2"
 hashbrown = { version = "0.13", default-features = false }
 hdrhistogram = "7"
 hex = "0.4"
-io-uring = { version = "0.5", optional = true }
+io-uring = { version = "0.6", optional = true }
 libc = "0.2"
 log = "0.4"
 metrics = { version = "0.20", optional = true }
@@ -48,7 +48,7 @@ rand = { version = "0.8", features = ["small_rng"] }
 serde = { version = "1", features = ["derive"] }
 signal-hook = { version = "0.3" }
 slab = "0.4"
-socket2 = { version = "0.4", features = ["all"] }
+socket2 = { version = "0.5", features = ["all"] }
 time = { version = "0.3", features = ["formatting"] }
 tinytemplate = "1"
 

aquatic_udp/src/workers/socket/uring/buf_ring.rs

@@ -743,7 +743,7 @@ impl RawBufRing {
         // Future: move to separate public function so other buf_ring implementations
         // can register, and unregister, the same way.
 
-        let res = CurrentRing::with(|ring| {
+        let res = CurrentRing::with(|ring| unsafe {
             ring.submitter()
                 .register_buf_ring(self.ring_addr as _, self.ring_entries(), bgid)
         });

aquatic_udp_load_test/Cargo.toml

@@ -28,7 +28,7 @@ mio = { version = "0.8", features = ["net", "os-poll"] }
 rand_distr = "0.4"
 rand = { version = "0.8", features = ["small_rng"] }
 serde = { version = "1", features = ["derive"] }
-socket2 = { version = "0.4", features = ["all"] }
+socket2 = { version = "0.5", features = ["all"] }
 
 [dev-dependencies]
 quickcheck = "1"

aquatic_ws/Cargo.toml

@@ -27,11 +27,11 @@ aquatic_toml_config.workspace = true
 aquatic_ws_protocol.workspace = true
 
 anyhow = "1"
-async-tungstenite = "0.19"
+async-tungstenite = "0.20"
 cfg-if = "1"
 futures = "0.3"
 futures-lite = "1"
-futures-rustls = "0.22"
+futures-rustls = "0.24"
 glommio = "0.8"
 hashbrown = { version = "0.13", features = ["serde"] }
 httparse = "1"
@@ -41,12 +41,12 @@ metrics-exporter-prometheus = { version = "0.11", optional = true, default-featu
 mimalloc = { version = "0.1", default-features = false }
 privdrop = "0.5"
 rand = { version = "0.8", features = ["small_rng"] }
-rustls = "0.20"
+rustls = "0.21"
 rustls-pemfile = "1"
 serde = { version = "1", features = ["derive"] }
 signal-hook = { version = "0.3" }
 slab = "0.4"
-socket2 = { version = "0.4", features = ["all"] }
+socket2 = { version = "0.5", features = ["all"] }
 tungstenite = "0.18"
 
 [dev-dependencies]

aquatic_ws_load_test/Cargo.toml

@@ -19,15 +19,15 @@ aquatic_toml_config.workspace = true
 aquatic_ws_protocol.workspace = true
 
 anyhow = "1"
-async-tungstenite = "0.19"
+async-tungstenite = "0.20"
 futures = "0.3"
-futures-rustls = "0.22"
+futures-rustls = "0.24"
 glommio = "0.8"
 log = "0.4"
 mimalloc = { version = "0.1", default-features = false }
 rand = { version = "0.8", features = ["small_rng"] }
 rand_distr = "0.4"
-rustls = { version = "0.20", default-features = false, features = ["dangerous_configuration"] }
+rustls = { version = "0.21", default-features = false, features = ["dangerous_configuration"] }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 tungstenite = "0.18"

aquatic_ws_protocol/Cargo.toml

@@ -30,6 +30,6 @@ simd-json = { version = "0.6", features = ["allow-non-simd"] }
 tungstenite = "0.18"
 
 [dev-dependencies]
-criterion = "0.3"
+criterion = "0.4"
 quickcheck = "1"
 quickcheck_macros = "1"

scripts/run-aquatic-http-private.sh

@@ -1,5 +0,0 @@
-#!/bin/sh 
-
-. ./scripts/env-native-cpu-without-avx-512
-
-cargo run --profile "release-debug" -p aquatic_http_private -- $@