server-core: new crate

Move the ore server feature to its own crate. No logic changes.

Cargo.toml

@@ -65,6 +65,7 @@ members = [
     "src/s3-datagen",
     "src/secrets",
     "src/segment",
+    "src/server-core",
     "src/service",
     "src/ssh-util",
     "src/sql",

src/balancerd/Cargo.toml

@@ -18,7 +18,8 @@ jsonwebtoken = "8.2.0"
 mz-build-info = { path = "../build-info" }
 mz-frontegg-auth = { path = "../frontegg-auth" }
 mz-orchestrator-tracing = { path = "../orchestrator-tracing" }
-mz-ore = { path = "../ore", features = ["server"] }
+mz-ore = { path = "../ore", default-features = false }
+mz-server-core = { path = "../server-core" }
 mz-pgwire-common = { path = "../pgwire-common" }
 num_cpus = "1.14.0"
 openssl = { version = "0.10.48", features = ["vendored"] }
@@ -31,9 +32,5 @@ tracing = "0.1.37"
 uuid = "1.2.2"
 workspace-hack = { version = "0.0.0", path = "../workspace-hack" }
 
-[features]
-default = ["tokio-console"]
-tokio-console = ["mz-ore/tokio-console"]
-
 [package.metadata.cargo-udeps.ignore]
 normal = ["workspace-hack"]

src/balancerd/src/lib.rs

@@ -98,12 +98,12 @@ use mz_frontegg_auth::Authentication as FronteggAuthentication;
 use mz_ore::metric;
 use mz_ore::metrics::{ComputedGauge, MetricsRegistry};
 use mz_ore::netio::AsyncReady;
-use mz_ore::server::{listen, TlsCertConfig, TlsConfig, TlsMode};
 use mz_ore::task::JoinSetExt;
 use mz_pgwire_common::{
     decode_startup, Conn, ErrorResponse, FrontendMessage, FrontendStartupMessage,
     ACCEPT_SSL_ENCRYPTION, REJECT_ENCRYPTION, VERSION_3,
 };
+use mz_server_core::{listen, TlsCertConfig, TlsConfig, TlsMode};
 use openssl::ssl::{NameType, Ssl, SslContext};
 use semver::Version;
 use tokio::io::{self, AsyncRead, AsyncWrite, AsyncWriteExt};
@@ -214,7 +214,7 @@ impl BalancerService {
                 tls: pgwire_tls,
             };
             set.spawn_named(|| "pgwire_stream", async move {
-                mz_ore::server::serve(pgwire_stream, pgwire).await;
+                mz_server_core::serve(pgwire_stream, pgwire).await;
             });
         }
         {
@@ -223,7 +223,7 @@ impl BalancerService {
                 resolve_template: Arc::from(self.cfg.https_addr_template),
             };
             set.spawn_named(|| "https_stream", async move {
-                mz_ore::server::serve(https_stream, https).await;
+                mz_server_core::serve(https_stream, https).await;
             });
         }
 
@@ -339,10 +339,10 @@ impl PgwireBalancer {
     }
 }
 
-impl mz_ore::server::Server for PgwireBalancer {
+impl mz_server_core::Server for PgwireBalancer {
     const NAME: &'static str = "pgwire_balancer";
 
-    fn handle_connection(&self, conn: TcpStream) -> mz_ore::server::ConnectionHandler {
+    fn handle_connection(&self, conn: TcpStream) -> mz_server_core::ConnectionHandler {
         let tls = self.tls.clone();
         let resolver = Arc::clone(&self.resolver);
         Box::pin(async move {
@@ -418,10 +418,10 @@ struct HttpsBalancer {
     // todo: metrics
 }
 
-impl mz_ore::server::Server for HttpsBalancer {
+impl mz_server_core::Server for HttpsBalancer {
     const NAME: &'static str = "https_balancer";
 
-    fn handle_connection(&self, conn: TcpStream) -> mz_ore::server::ConnectionHandler {
+    fn handle_connection(&self, conn: TcpStream) -> mz_server_core::ConnectionHandler {
         let tls_context = Arc::clone(&self.tls);
         let resolve_template = Arc::clone(&self.resolve_template);
         Box::pin(async move {

src/balancerd/src/service.rs

@@ -17,7 +17,7 @@ use mz_balancerd::{
 };
 use mz_frontegg_auth::{Authentication, AuthenticationConfig};
 use mz_ore::metrics::MetricsRegistry;
-use mz_ore::server::TlsCliArgs;
+use mz_server_core::TlsCliArgs;
 use tracing::info;
 
 #[derive(Debug, clap::Parser)]

src/environmentd/Cargo.toml

@@ -61,6 +61,7 @@ mz-prof = { path = "../prof" }
 mz-repr = { path = "../repr" }
 mz-secrets = { path = "../secrets" }
 mz-segment = { path = "../segment" }
+mz-server-core = { path = "../server-core" }
 mz-service = { path = "../service" }
 mz-sql = { path = "../sql" }
 mz-stash = { path = "../stash" }

src/environmentd/src/bin/environmentd/main.rs

@@ -112,7 +112,6 @@ use mz_ore::cli::{self, CliConfig, KeyValueArg};
 use mz_ore::error::ErrorExt;
 use mz_ore::metrics::MetricsRegistry;
 use mz_ore::now::SYSTEM_TIME;
-use mz_ore::server::TlsCliArgs;
 use mz_ore::task::RuntimeExt;
 use mz_ore::{halt, metric};
 use mz_persist_client::cache::PersistClientCache;
@@ -122,6 +121,7 @@ use mz_persist_client::rpc::{
 };
 use mz_persist_client::PersistLocation;
 use mz_secrets::SecretsController;
+use mz_server_core::TlsCliArgs;
 use mz_service::emit_boot_diagnostics;
 use mz_service::secrets::{SecretsControllerKind, SecretsReaderCliArgs};
 use mz_sql::catalog::EnvironmentId;

src/environmentd/src/http.rs

@@ -43,9 +43,9 @@ use mz_frontegg_auth::{
 use mz_http_util::DynamicFilterTarget;
 use mz_ore::cast::u64_to_usize;
 use mz_ore::metrics::MetricsRegistry;
-use mz_ore::server::{ConnectionHandler, Server};
 use mz_ore::str::StrExt;
 use mz_repr::user::ExternalUserMetadata;
+use mz_server_core::{ConnectionHandler, Server};
 use mz_sql::session::user::{
     User, HTTP_DEFAULT_USER, SUPPORT_USER, SUPPORT_USER_NAME, SYSTEM_USER, SYSTEM_USER_NAME,
 };

src/environmentd/src/lib.rs

@@ -103,11 +103,11 @@ use mz_frontegg_auth::Authentication as FronteggAuthentication;
 use mz_ore::future::OreFutureExt;
 use mz_ore::metrics::MetricsRegistry;
 use mz_ore::now::NowFn;
-use mz_ore::server::{ConnectionStream, ListenerHandle, TlsCertConfig};
 use mz_ore::task;
 use mz_ore::tracing::TracingHandle;
 use mz_persist_client::usage::StorageUsageClient;
 use mz_secrets::SecretsController;
+use mz_server_core::{ConnectionStream, ListenerHandle, TlsCertConfig};
 use mz_sql::catalog::EnvironmentId;
 use mz_sql::session::vars::ConnectionCounter;
 use mz_storage_types::connections::ConnectionContext;
@@ -273,12 +273,12 @@ impl Listeners {
             internal_http_listen_addr,
         }: ListenersConfig,
     ) -> Result<Listeners, anyhow::Error> {
-        let sql = mz_ore::server::listen(sql_listen_addr).await?;
-        let http = mz_ore::server::listen(http_listen_addr).await?;
-        let balancer_sql = mz_ore::server::listen(balancer_sql_listen_addr).await?;
-        let balancer_http = mz_ore::server::listen(balancer_http_listen_addr).await?;
-        let internal_sql = mz_ore::server::listen(internal_sql_listen_addr).await?;
-        let internal_http = mz_ore::server::listen(internal_http_listen_addr).await?;
+        let sql = mz_server_core::listen(sql_listen_addr).await?;
+        let http = mz_server_core::listen(http_listen_addr).await?;
+        let balancer_sql = mz_server_core::listen(balancer_sql_listen_addr).await?;
+        let balancer_http = mz_server_core::listen(balancer_http_listen_addr).await?;
+        let internal_sql = mz_server_core::listen(internal_sql_listen_addr).await?;
+        let internal_http = mz_server_core::listen(internal_http_listen_addr).await?;
         Ok(Listeners {
             sql,
             http,
@@ -326,9 +326,9 @@ impl Listeners {
             None => (None, None),
             Some(tls_config) => {
                 let context = tls_config.context()?;
-                let pgwire_tls = mz_ore::server::TlsConfig {
+                let pgwire_tls = mz_server_core::TlsConfig {
                     context: context.clone(),
-                    mode: mz_ore::server::TlsMode::Require,
+                    mode: mz_server_core::TlsMode::Require,
                 };
                 let http_tls = http::TlsConfig {
                     context,
@@ -360,7 +360,7 @@ impl Listeners {
                 ready_to_promote: ready_to_promote_rx,
                 internal_console_redirect_url: config.internal_console_redirect_url,
             });
-            mz_ore::server::serve(internal_http_conns, internal_http_server)
+            mz_server_core::serve(internal_http_conns, internal_http_server)
         });
 
         let mut openable_adapter_storage = mz_catalog::stash_backed_catalog_state(StashConfig {
@@ -527,7 +527,7 @@ impl Listeners {
                 internal: false,
                 active_connection_count: Arc::clone(&active_connection_count),
             });
-            mz_ore::server::serve(sql_conns, sql_server)
+            mz_server_core::serve(sql_conns, sql_server)
         });
 
         // Launch internal SQL server.
@@ -540,7 +540,7 @@ impl Listeners {
                     //
                     // TODO(benesch): migrate all internal applications to TLS and
                     // remove `TlsMode::Allow`.
-                    pgwire_tls.mode = mz_ore::server::TlsMode::Allow;
+                    pgwire_tls.mode = mz_server_core::TlsMode::Allow;
                     pgwire_tls
                 }),
                 adapter_client: adapter_client.clone(),
@@ -549,7 +549,7 @@ impl Listeners {
                 internal: true,
                 active_connection_count: Arc::clone(&active_connection_count),
             });
-            mz_ore::server::serve(internal_sql_conns, internal_sql_server)
+            mz_server_core::serve(internal_sql_conns, internal_sql_server)
         });
 
         // Launch HTTP server.
@@ -566,7 +566,7 @@ impl Listeners {
                     .unwrap_or(http::WEBHOOK_CONCURRENCY_LIMIT),
                 metrics: http_metrics.clone(),
             });
-            mz_ore::server::serve(http_conns, http_server)
+            mz_server_core::serve(http_conns, http_server)
         });
 
         // Launch HTTP server exposed to balancers
@@ -583,7 +583,7 @@ impl Listeners {
                     .unwrap_or(http::WEBHOOK_CONCURRENCY_LIMIT),
                 metrics: http_metrics,
             });
-            mz_ore::server::serve(balancer_http_conns, balancer_http_server)
+            mz_server_core::serve(balancer_http_conns, balancer_http_server)
         });
 
         // Launch SQL server exposed to balancers
@@ -596,7 +596,7 @@ impl Listeners {
                 internal: false,
                 active_connection_count: Arc::clone(&active_connection_count),
             });
-            mz_ore::server::serve(balancer_sql_conns, balancer_sql_server)
+            mz_server_core::serve(balancer_sql_conns, balancer_sql_server)
         });
 
         // Start telemetry reporting loop.

src/environmentd/tests/util.rs

@@ -91,7 +91,6 @@ use mz_orchestrator_process::{ProcessOrchestrator, ProcessOrchestratorConfig};
 use mz_ore::metrics::MetricsRegistry;
 use mz_ore::now::{EpochMillis, NowFn, SYSTEM_TIME};
 use mz_ore::retry::Retry;
-use mz_ore::server::TlsCertConfig;
 use mz_ore::task;
 use mz_ore::tracing::{
     OpenTelemetryConfig, StderrLogConfig, StderrLogFormat, TracingConfig, TracingGuard,
@@ -102,6 +101,7 @@ use mz_persist_client::cfg::{PersistConfig, PersistParameters};
 use mz_persist_client::rpc::PersistGrpcPubSubServer;
 use mz_persist_client::PersistLocation;
 use mz_secrets::SecretsController;
+use mz_server_core::TlsCertConfig;
 use mz_sql::catalog::EnvironmentId;
 use mz_stash_types::metrics::Metrics as StashMetrics;
 use mz_storage_types::connections::ConnectionContext;

src/ore/Cargo.toml

@@ -35,7 +35,6 @@ smallvec = { version = "1.10.0", optional = true }
 stacker = { version = "0.1.15", optional = true }
 sentry = { version = "0.29.1", optional = true, features = ["debug-images"] }
 serde = { version = "1.0.152", features = ["derive"], optional = true }
-socket2 = { version = "0.4.7", optional = true }
 tokio = { version = "1.32.0", features = [
   "io-util",
   "net",
@@ -44,7 +43,6 @@ tokio = { version = "1.32.0", features = [
   "time",
 ], optional = true }
 tokio-openssl = { version = "0.6.3", optional = true }
-tokio-stream = { version = "0.1.11", features = ["net"], optional = true }
 # TODO(guswynn): determine, when, if ever, we can remove `tracing-log`
 # The `tracing-log` feature here is load-bearing: While our busiest-logging dependency (`rdkafka`) is now hooked-up
 # to use `tracing`, we cannot remove this feature until we guarantee no dependencies log using the `log` crate, for
@@ -98,7 +96,6 @@ async = [
 ]
 bytes_ = ["bytes", "smallvec", "smallvec/const_generics"]
 network = ["async", "bytes", "hyper", "smallvec", "tonic", "tracing"]
-server = ["openssl", "socket2", "tokio-stream"]
 tracing_ = [
   "anyhow",
   "atty",

src/ore/src/lib.rs

@@ -137,8 +137,6 @@ pub mod result;
 pub mod retry;
 #[cfg(feature = "serde")]
 pub mod serde;
-#[cfg(feature = "server")]
-pub mod server;
 #[cfg_attr(nightly_doc_features, doc(cfg(feature = "stack")))]
 #[cfg(feature = "stack")]
 pub mod stack;

src/pgwire-common/Cargo.toml

@@ -12,7 +12,8 @@ async-trait = "0.1.68"
 byteorder = "1.4.3"
 bytes = "1.3.0"
 bytesize = "1.1.0"
-mz-ore = { path = "../ore", features = ["network", "server"] }
+mz-ore = { path = "../ore", features = ["network"] }
+mz-server-core = { path = "../server-core" }
 tokio = "1.24.2"
 tokio-openssl = "0.6.3"
 tokio-postgres = { version = "0.7.8" }

src/pgwire-common/src/conn.rs

@@ -12,7 +12,7 @@ use std::task::{Context, Poll};
 
 use async_trait::async_trait;
 use mz_ore::netio::AsyncReady;
-use mz_ore::server::TlsMode;
+use mz_server_core::TlsMode;
 use tokio::io::{self, AsyncRead, AsyncWrite, Interest, ReadBuf, Ready};
 use tokio_openssl::SslStream;
 use tokio_postgres::error::SqlState;

src/pgwire/Cargo.toml

@@ -17,11 +17,12 @@ itertools = "0.10.5"
 mz-adapter = { path = "../adapter" }
 mz-expr = { path = "../expr" }
 mz-frontegg-auth = { path = "../frontegg-auth" }
-mz-ore = { path = "../ore", features = ["tracing_", "server"] }
+mz-ore = { path = "../ore", features = ["tracing_"] }
 mz-pgcopy = { path = "../pgcopy" }
 mz-pgrepr = { path = "../pgrepr" }
 mz-pgwire-common = { path = "../pgwire-common" }
 mz-repr = { path = "../repr" }
+mz-server-core = { path = "../server-core" }
 mz-sql = { path = "../sql" }
 openssl = { version = "0.10.48", features = ["vendored"] }
 postgres = { version = "0.19.5" }