From 0dcba14572b05dde9150cff01fc853669de35618 Mon Sep 17 00:00:00 2001 From: Zyy Date: Thu, 1 Oct 2026 16:00:07 +0800 Subject: [PATCH] refactor(bootstrap): separate startup modes and plugin wiring Co-authored-by: Codex --- backend/apps/gateway/src/bootstrap.rs | 463 ------------------ backend/apps/gateway/src/bootstrap/command.rs | 134 +++++ backend/apps/gateway/src/bootstrap/config.rs | 48 ++ backend/apps/gateway/src/bootstrap/mod.rs | 42 ++ backend/apps/gateway/src/bootstrap/plugins.rs | 108 ++++ backend/apps/gateway/src/bootstrap/server.rs | 164 +++++++ backend/apps/gateway/src/bootstrap/startup.rs | 94 ++++ .../apps/gateway/tests/bootstrap/command.rs | 49 ++ .../{bootstrap.rs => bootstrap/config.rs} | 201 +------- backend/apps/gateway/tests/bootstrap/mod.rs | 3 + .../apps/gateway/tests/bootstrap/server.rs | 146 ++++++ backend/apps/gateway/tests/main.rs | 12 +- 12 files changed, 802 insertions(+), 662 deletions(-) delete mode 100644 backend/apps/gateway/src/bootstrap.rs create mode 100644 backend/apps/gateway/src/bootstrap/command.rs create mode 100644 backend/apps/gateway/src/bootstrap/config.rs create mode 100644 backend/apps/gateway/src/bootstrap/mod.rs create mode 100644 backend/apps/gateway/src/bootstrap/plugins.rs create mode 100644 backend/apps/gateway/src/bootstrap/server.rs create mode 100644 backend/apps/gateway/src/bootstrap/startup.rs create mode 100644 backend/apps/gateway/tests/bootstrap/command.rs rename backend/apps/gateway/tests/{bootstrap.rs => bootstrap/config.rs} (68%) create mode 100644 backend/apps/gateway/tests/bootstrap/mod.rs create mode 100644 backend/apps/gateway/tests/bootstrap/server.rs diff --git a/backend/apps/gateway/src/bootstrap.rs b/backend/apps/gateway/src/bootstrap.rs deleted file mode 100644 index 84f86d686..000000000 --- a/backend/apps/gateway/src/bootstrap.rs +++ /dev/null @@ -1,463 +0,0 @@ -//! 网关唯一组合根：加载包级配置并连接各 Bundle。 - -use gateway_core::engine::provider::{ProviderRegistry, RegistryError}; -use gateway_host::{ConfigError, HostConfig, LoadableConfig}; -use serde::Deserialize; - -const CONFIG_SCHEMA_VERSION: u32 = 1; - -/// 顶层配置只组合各包拥有的配置段，不解释任何业务字段。 -/// 各启动配置忽略未知字段，允许升级时保留旧配置；已知字段仍按类型和业务约束校验。 -#[derive(Debug, Deserialize)] -pub struct GatewayConfig { - schema_version: u32, - host: HostConfig, - store: gateway_store::StoreConfig, - admin: gateway_admin::AdminConfig, - #[serde(default)] - client: gateway_admin::ClientConfig, - api: gateway_api::ApiConfig, - #[serde(default)] - openai: provider_openai::OpenAiConfig, -} - -impl LoadableConfig for GatewayConfig { - const EXTERNAL_SECTIONS: &'static [&'static str] = &["services"]; - - fn resolve_and_validate(&mut self, source_dir: &std::path::Path) -> Result<(), ConfigError> { - if self.schema_version != CONFIG_SCHEMA_VERSION { - return Err(ConfigError::InvalidField("schema_version")); - } - self.api - .resolve_and_validate(source_dir) - .map_err(|_| ConfigError::InvalidField("api"))?; - self.host - .resolve_and_validate(source_dir, &self.api.asset_directory)?; - let runtime_data_dir = self.host.runtime_data_dir().to_path_buf(); - self.store - .resolve_and_validate(&runtime_data_dir) - .map_err(|_| ConfigError::InvalidField("store"))?; - self.admin - .resolve_and_validate(source_dir) - .map_err(|_| ConfigError::InvalidField("admin"))?; - self.client - .resolve_and_validate(source_dir) - .map_err(|_| ConfigError::InvalidField("client"))?; - self.openai - .resolve_and_validate(&runtime_data_dir) - .map_err(|_| ConfigError::InvalidField("openai"))?; - Ok(()) - } -} - -/// 按冻结顺序初始化全部 Bundle，并把进程阻塞权交给 Host。 -pub async fn run() -> Result<(), BootstrapError> { - launch(None).await.map(|_| ()) -} - -/// 这里只接收插件命名空间内的参数，不转交宿主配置或其他启动 secret。 -pub struct PluginCommand { - pub instance_id: Option, - pub name: Option, - pub arguments: Vec, -} - -impl PluginCommand { - fn is_help(&self) -> bool { - self.name.is_none() - || self - .arguments - .iter() - .any(|value| matches!(value.as_str(), "--help" | "-h")) - } -} - -pub async fn plugin_command( - command: PluginCommand, -) -> Result { - launch(Some(command)).await -} - -async fn launch( - command: Option, -) -> Result { - macro_rules! or_shutdown { - ($runtime:expr, $result:expr) => { - match $result { - Ok(value) => value, - Err(error) => { - $runtime.shutdown().await; - return Err(error.into()); - } - } - }; - } - - let config = gateway_host::load_config::()?; - let GatewayConfig { - schema_version: _, - host, - store, - admin, - client, - api, - openai, - } = config; - - let timezone = host.timezone; - let store = store.with_timezone(timezone); - let plugin_cache = host.runtime_data_dir().join("plugins"); - let host = if command.is_some() { - gateway_host::initialize_command_line(host).await? - } else { - gateway_host::initialize(host).await? - }; - host.report_startup_ready("Host"); - let help = command.as_ref().is_some_and(PluginCommand::is_help); - let mut store = if help { - gateway_store::initialize_read_only(store).await? - } else if command.is_some() { - gateway_store::initialize_command_line(store).await? - } else { - gateway_store::initialize(store).await? - }; - host.report_startup_ready("Store"); - let version = host - .system_operations() - .version() - .await - .map_err(|_| ConfigError::InvalidField("host.version"))? - .version; - let plugin_http = std::sync::Arc::new(gateway_host::outbound::HttpClient::new()?); - if let Some(command) = command.as_ref().filter(|_| help) { - let runtime = prepare_plugins(&store, plugin_cache, &version, plugin_http.clone())?; - let commands = or_shutdown!(runtime, runtime.prepare_command_line().await); - let stdout = commands.help(command.instance_id.as_deref(), command.name.as_deref()); - commands.shutdown().await; - runtime.shutdown().await; - return Ok(gateway_plugin_runtime::PluginCommandOutput { - stdout: stdout?, - stderr: String::new(), - exit_code: 0, - saved_accounts: 0, - }); - } - let provider_ports = store.provider_ports(); - let mut openai = - provider_openai::initialize(openai.with_timezone(timezone), provider_ports.clone()).await?; - host.report_startup_ready("OpenAI Provider"); - let mut xai = provider_xai::initialize(provider_ports).await?; - host.report_startup_ready("xAI Provider"); - let providers = ProviderRegistry::new([openai.core_provider(), xai.core_provider()])?; - let plugin_runtime = prepare_plugins(&store, plugin_cache, &version, plugin_http.clone())?; - let admin_providers = gateway_admin::ports::provider::ProviderAdminRegistry::new([ - openai.admin_provider(), - xai.admin_provider(), - ]) - .map_err(|_| gateway_admin::model::AdminError::invalid("Provider 管理身份冲突"))?; - if let Some(command) = command { - let instance_id = or_shutdown!( - plugin_runtime, - command - .instance_id - .as_deref() - .ok_or_else(|| gateway_admin::model::AdminError::invalid("插件实例 ID 缺失")) - ); - let name = or_shutdown!( - plugin_runtime, - command - .name - .as_deref() - .ok_or_else(|| gateway_admin::model::AdminError::invalid("插件命令名缺失")) - ); - let core = gateway_core::prepare_command_plane( - store.core_ports(), - providers.clone(), - Some(plugin_runtime.clone()), - Some(plugin_runtime.observer_registry()), - Some(plugin_runtime.policy_registry()), - Some(plugin_runtime.middleware_registry()), - ); - let accounts = gateway_admin::initialize_plugin_accounts( - admin_providers.clone(), - store.admin_ports().accounts(), - core.snapshot_control(), - ); - or_shutdown!(plugin_runtime, plugin_runtime.bind_account_ports(&accounts)); - let keys = gateway_admin::initialize_plugin_client_keys( - admin_providers.clone(), - store.admin_ports().client_keys(), - core.snapshot_control(), - ); - or_shutdown!(plugin_runtime, plugin_runtime.bind_client_key_ports(&keys)); - let plugin_resources = gateway_admin::initialize_plugin_resources( - store.admin_ports().plugin_resources(), - core.snapshot_control(), - ); - or_shutdown!( - plugin_runtime, - plugin_runtime.bind_resource_ports(&plugin_resources) - ); - let core = or_shutdown!(plugin_runtime, core.activate().await); - let nested_models = core.nested_model_execution_port(); - let affinity_lookup = core.affinity_lookup_port(); - or_shutdown!( - plugin_runtime, - plugin_runtime.bind_model_ports(&nested_models, &affinity_lookup) - ); - let settings = gateway_admin::initialize_settings( - store.admin_ports().settings(), - core.snapshot_control(), - admin_providers, - std::sync::Arc::new(gateway_host::pricing::ModelsDevPricing), - ); - let mut services = gateway_admin::service::Registry::new({ - let snapshots = core.snapshots(); - let middleware = plugin_runtime.middleware_registry(); - std::sync::Arc::new(move || { - let snapshot = snapshots.snapshot_for_diagnostics()?; - middleware.resolve(snapshot.extensions()?) - }) - }); - or_shutdown!(plugin_runtime, services.register_settings(&settings)); - let services = std::sync::Arc::new(services); - or_shutdown!(plugin_runtime, plugin_runtime.bind_services(&services)); - let commands = or_shutdown!(plugin_runtime, plugin_runtime.prepare_command_line().await); - or_shutdown!(plugin_runtime, store.start_command_line_writes()); - let result = commands - .execute_cancellable(instance_id, name, &command.arguments, &host.cancellation()) - .await; - commands.shutdown().await; - plugin_runtime.shutdown().await; - store.shutdown_command_line_writes().await?; - return Ok(result?); - } - let core = gateway_core::prepare( - store.core_ports(), - providers, - Some(plugin_runtime.clone()), - Some(plugin_runtime.observer_registry()), - Some(plugin_runtime.policy_registry()), - Some(plugin_runtime.middleware_registry()), - Some(plugin_runtime.frontend_authentication_registry()), - ); - let plugin_accounts = gateway_admin::initialize_plugin_accounts( - admin_providers.clone(), - store.admin_ports().accounts(), - core.snapshot_control(), - ); - or_shutdown!( - plugin_runtime, - plugin_runtime.bind_account_ports(&plugin_accounts) - ); - let plugin_keys = gateway_admin::initialize_plugin_client_keys( - admin_providers.clone(), - store.admin_ports().client_keys(), - core.snapshot_control(), - ); - or_shutdown!( - plugin_runtime, - plugin_runtime.bind_client_key_ports(&plugin_keys) - ); - let plugin_resources = gateway_admin::initialize_plugin_resources( - store.admin_ports().plugin_resources(), - core.snapshot_control(), - ); - or_shutdown!( - plugin_runtime, - plugin_runtime.bind_resource_ports(&plugin_resources) - ); - let mut core = or_shutdown!(plugin_runtime, core.activate().await); - let nested_models = core.nested_model_execution_port(); - let affinity_lookup = core.affinity_lookup_port(); - or_shutdown!( - plugin_runtime, - plugin_runtime.bind_model_ports(&nested_models, &affinity_lookup) - ); - host.report_startup_ready("Core"); - let host_version = or_shutdown!( - plugin_runtime, - version - .trim_start_matches('v') - .parse() - .map_err(|_| ConfigError::InvalidField("host.version")) - ); - let plugin_inspector = std::sync::Arc::new(gateway_plugin_runtime::PackageInspector::new( - gateway_plugin_runtime::PackageLimits::default(), - host_version, - )); - let plugin_distribution = std::sync::Arc::new(or_shutdown!( - plugin_runtime, - gateway_host::plugin_distribution::HttpPluginDistribution::new(plugin_http) - )); - let mut admin = or_shutdown!( - plugin_runtime, - gateway_admin::initialize_with_plugin_accounts( - admin, - client, - store.admin_ports(), - gateway_admin::AdminRuntimePorts { - timezone, - service_middleware: { - let snapshots = core.snapshots(); - let middleware = plugin_runtime.middleware_registry(); - std::sync::Arc::new(move || { - let snapshot = snapshots.snapshot_for_diagnostics()?; - middleware.resolve(snapshot.extensions()?) - }) - }, - providers: admin_providers, - plugin_preparation: plugin_runtime.clone(), - plugin_management: plugin_runtime.clone(), - published_snapshot: core.snapshots(), - plugin_inspector, - plugin_distribution, - pricing_source: std::sync::Arc::new(gateway_host::pricing::ModelsDevPricing), - snapshot: core.snapshot_control(), - account_probe: core.account_probe(), - proxy_probe: host.proxy_probe(provider_openai::build_reqwest_client_with_custom_ca), - client_distribution: host.client_distribution_resolver(), - system: host.system_operations(), - client_key_verifier: core.client_key_verifier(), - }, - plugin_accounts, - ) - .await - ); - let official_files = host.official_plugin_release_files(); - let official_identity = host.official_plugin_release_identity(); - let official_import = or_shutdown!( - plugin_runtime, - admin - .services() - .plugins() - .import_official_release( - official_files.as_ref(), - &official_identity, - &gateway_admin::model::MutationContext { - actor: gateway_admin::model::MutationActor::System, - request_id: "startup-official-plugin-import".to_owned(), - }, - ) - .await - ); - if official_import.artifacts > 0 { - host.report_startup_ready("Official plugins"); - } - host.report_startup_ready("Admin"); - - let mut probes = store.health_probes(); - probes.extend(core.health_probes()); - probes.push(host.logging_health_probe()); - let api = or_shutdown!( - plugin_runtime, - gateway_api::initialize( - api, - core.execution_service(), - admin.services(), - probes, - host.worker_health(), - host.connection_lifecycle(), - ) - ) - .with_middleware({ - let snapshots = core.snapshots(); - let middleware = plugin_runtime.middleware_registry(); - move |snapshot| { - if let Some(snapshot) = snapshot { - return middleware.resolve(snapshot.extensions()?); - } - // 数据面未就绪时，管理与诊断路由仍可使用发布候选中的插件。 - let diagnostic = snapshots.snapshot_for_diagnostics()?; - middleware.resolve(diagnostic.extensions()?) - } - }); - let http_dispatcher = api.dispatcher(); - or_shutdown!(plugin_runtime, plugin_runtime.bind_http(&http_dispatcher)); - host.report_startup_ready("API"); - - or_shutdown!( - plugin_runtime, - plugin_runtime.bind_services(&admin.services().public_services()) - ); - - let mut plan = store.take_worker_contributions(); - plan.push(gateway_host::retention::worker(store.retention())?); - plan.extend(core.take_worker_contributions()); - plan.extend(openai.take_worker_contributions()); - plan.extend(xai.take_worker_contributions()); - plan.extend(admin.take_worker_contributions()); - plan.push(plugin_runtime.maintenance_worker(core.snapshots())?); - or_shutdown!( - plugin_runtime, - host.start_workers(plan, store.worker_leader_lease()) - ); - host.report_startup_ready("Workers"); - let served = host.serve(api.router()).await; - plugin_runtime.shutdown().await; - served?; - Ok(gateway_plugin_runtime::PluginCommandOutput { - stdout: String::new(), - stderr: String::new(), - exit_code: 0, - saved_accounts: 0, - }) -} - -fn prepare_plugins( - store: &gateway_store::StoreBundle, - plugin_cache: std::path::PathBuf, - version: &str, - http: std::sync::Arc, -) -> Result, BootstrapError> { - Ok(std::sync::Arc::new( - gateway_plugin_runtime::PluginRuntime::new( - store.admin_ports().plugins(), - store.admin_ports().plugin_state(), - gateway_plugin_runtime::PluginRuntimeConfig { - cache_directory: plugin_cache, - host_version: version - .trim_start_matches('v') - .parse() - .map_err(|_| ConfigError::InvalidField("host.version"))?, - package_limits: gateway_plugin_runtime::PackageLimits::default(), - rpc_limits: gateway_plugin_runtime::RpcLimits::default(), - restart_circuit: Default::default(), - }, - http, - std::sync::Arc::new(gateway_host::process::ProcessSupervisor::default()), - ) - .with_oauth_pending(store.provider_ports().oauth_pending()), - )) -} - -/// 组合根只保留包级错误分类，不展开内部实现或敏感配置。 -#[derive(Debug, thiserror::Error)] -pub enum BootstrapError { - #[error(transparent)] - Worker(#[from] gateway_core::task::WorkerDefinitionError), - #[error(transparent)] - PluginCommand(#[from] gateway_plugin_runtime::PluginCommandError), - #[error(transparent)] - Config(#[from] gateway_host::ConfigError), - #[error(transparent)] - Host(#[from] gateway_host::HostError), - #[error(transparent)] - Outbound(#[from] gateway_host::outbound::HttpError), - #[error(transparent)] - Store(#[from] gateway_store::StoreError), - #[error(transparent)] - CommandStoreDrain(#[from] gateway_store::CommandStoreDrainError), - #[error(transparent)] - OpenAi(#[from] provider_openai::OpenAiInitializeError), - #[error(transparent)] - Xai(#[from] provider_xai::XaiInitializeError), - #[error(transparent)] - Registry(#[from] RegistryError), - #[error(transparent)] - Core(#[from] gateway_core::CoreError), - #[error(transparent)] - Admin(#[from] gateway_admin::model::AdminError), - #[error(transparent)] - Api(#[from] gateway_api::ApiError), -} diff --git a/backend/apps/gateway/src/bootstrap/command.rs b/backend/apps/gateway/src/bootstrap/command.rs new file mode 100644 index 000000000..214844327 --- /dev/null +++ b/backend/apps/gateway/src/bootstrap/command.rs @@ -0,0 +1,134 @@ +//! 插件帮助只读启动；命令仅组装命令平面和必要写泵。 + +use super::{ + BootstrapError, GatewayConfig, + plugins::{ManagementPorts, or_shutdown, prepare_plugins, service_middleware}, + startup::{Environment, Mode, Providers}, +}; + +/// 这里只接收插件命名空间内的参数，不转交宿主配置或其他启动 secret。 +pub struct PluginCommand { + pub instance_id: Option, + pub name: Option, + pub arguments: Vec, +} + +impl PluginCommand { + fn is_help(&self) -> bool { + self.name.is_none() + || self + .arguments + .iter() + .any(|value| matches!(value.as_str(), "--help" | "-h")) + } +} + +pub async fn plugin_command( + command: PluginCommand, +) -> Result { + let config = gateway_host::load_config::()?; + if command.is_help() { + let environment = Environment::initialize(config.host, config.store, Mode::Help).await?; + help(command, environment).await + } else { + let environment = Environment::initialize(config.host, config.store, Mode::Command).await?; + let providers = Providers::initialize(config.openai, &environment).await?; + execute(command, environment, providers).await + } +} + +async fn help( + command: PluginCommand, + environment: Environment, +) -> Result { + let runtime = prepare_plugins( + &environment.store, + environment.plugin_cache, + &environment.version, + environment.plugin_http, + )?; + let commands = or_shutdown!(runtime, runtime.prepare_command_line().await); + let stdout = commands.help(command.instance_id.as_deref(), command.name.as_deref()); + commands.shutdown().await; + runtime.shutdown().await; + Ok(gateway_plugin_runtime::PluginCommandOutput { + stdout: stdout?, + stderr: String::new(), + exit_code: 0, + saved_accounts: 0, + }) +} + +async fn execute( + command: PluginCommand, + environment: Environment, + providers: Providers, +) -> Result { + let Environment { + host, + mut store, + plugin_cache, + version, + plugin_http, + .. + } = environment; + let plugin_runtime = prepare_plugins(&store, plugin_cache, &version, plugin_http)?; + let instance_id = or_shutdown!( + plugin_runtime, + command + .instance_id + .as_deref() + .ok_or_else(|| gateway_admin::model::AdminError::invalid("插件实例 ID 缺失")) + ); + let name = or_shutdown!( + plugin_runtime, + command + .name + .as_deref() + .ok_or_else(|| gateway_admin::model::AdminError::invalid("插件命令名缺失")) + ); + let core = gateway_core::prepare_command_plane( + store.core_ports(), + providers.core, + Some(plugin_runtime.clone()), + Some(plugin_runtime.observer_registry()), + Some(plugin_runtime.policy_registry()), + Some(plugin_runtime.middleware_registry()), + ); + let management = ManagementPorts::new(&store, providers.admin.clone(), core.snapshot_control()); + or_shutdown!(plugin_runtime, management.bind(&plugin_runtime)); + let core = or_shutdown!(plugin_runtime, core.activate().await); + let nested_models = core.nested_model_execution_port(); + let affinity_lookup = core.affinity_lookup_port(); + or_shutdown!( + plugin_runtime, + plugin_runtime.bind_model_ports(&nested_models, &affinity_lookup) + ); + let settings = gateway_admin::initialize_settings( + store.admin_ports().settings(), + core.snapshot_control(), + providers.admin, + std::sync::Arc::new(gateway_host::pricing::ModelsDevPricing), + ); + let mut services = gateway_admin::service::Registry::new(service_middleware( + core.snapshots(), + &plugin_runtime, + )); + or_shutdown!(plugin_runtime, services.register_settings(&settings)); + let services = std::sync::Arc::new(services); + or_shutdown!(plugin_runtime, plugin_runtime.bind_services(&services)); + let commands = or_shutdown!(plugin_runtime, plugin_runtime.prepare_command_line().await); + let result = async { + store.start_command_line_writes()?; + Ok::<_, BootstrapError>( + commands + .execute_cancellable(instance_id, name, &command.arguments, &host.cancellation()) + .await?, + ) + } + .await; + commands.shutdown().await; + plugin_runtime.shutdown().await; + store.shutdown_command_line_writes().await?; + result +} diff --git a/backend/apps/gateway/src/bootstrap/config.rs b/backend/apps/gateway/src/bootstrap/config.rs new file mode 100644 index 000000000..b3f0cc184 --- /dev/null +++ b/backend/apps/gateway/src/bootstrap/config.rs @@ -0,0 +1,48 @@ +use gateway_host::{ConfigError, HostConfig, LoadableConfig}; +use serde::Deserialize; + +const CONFIG_SCHEMA_VERSION: u32 = 1; + +/// 顶层配置只组合各包拥有的配置段，不解释任何业务字段。 +/// 各启动配置忽略未知字段，允许升级时保留旧配置；已知字段仍按类型和业务约束校验。 +#[derive(Debug, Deserialize)] +pub struct GatewayConfig { + schema_version: u32, + pub(super) host: HostConfig, + pub(super) store: gateway_store::StoreConfig, + pub(super) admin: gateway_admin::AdminConfig, + #[serde(default)] + pub(super) client: gateway_admin::ClientConfig, + pub(super) api: gateway_api::ApiConfig, + #[serde(default)] + pub(super) openai: provider_openai::OpenAiConfig, +} + +impl LoadableConfig for GatewayConfig { + const EXTERNAL_SECTIONS: &'static [&'static str] = &["services"]; + + fn resolve_and_validate(&mut self, source_dir: &std::path::Path) -> Result<(), ConfigError> { + if self.schema_version != CONFIG_SCHEMA_VERSION { + return Err(ConfigError::InvalidField("schema_version")); + } + self.api + .resolve_and_validate(source_dir) + .map_err(|_| ConfigError::InvalidField("api"))?; + self.host + .resolve_and_validate(source_dir, &self.api.asset_directory)?; + let runtime_data_dir = self.host.runtime_data_dir().to_path_buf(); + self.store + .resolve_and_validate(&runtime_data_dir) + .map_err(|_| ConfigError::InvalidField("store"))?; + self.admin + .resolve_and_validate(source_dir) + .map_err(|_| ConfigError::InvalidField("admin"))?; + self.client + .resolve_and_validate(source_dir) + .map_err(|_| ConfigError::InvalidField("client"))?; + self.openai + .resolve_and_validate(&runtime_data_dir) + .map_err(|_| ConfigError::InvalidField("openai"))?; + Ok(()) + } +} diff --git a/backend/apps/gateway/src/bootstrap/mod.rs b/backend/apps/gateway/src/bootstrap/mod.rs new file mode 100644 index 000000000..e43240254 --- /dev/null +++ b/backend/apps/gateway/src/bootstrap/mod.rs @@ -0,0 +1,42 @@ +//! 网关唯一组合根：按启动模式连接各 Bundle。 + +mod command; +mod config; +mod plugins; +mod server; +mod startup; + +pub use command::{PluginCommand, plugin_command}; +pub use config::GatewayConfig; +pub use server::run; + +/// 组合根只保留包级错误分类，不展开内部实现或敏感配置。 +#[derive(Debug, thiserror::Error)] +pub enum BootstrapError { + #[error(transparent)] + Worker(#[from] gateway_core::task::WorkerDefinitionError), + #[error(transparent)] + PluginCommand(#[from] gateway_plugin_runtime::PluginCommandError), + #[error(transparent)] + Config(#[from] gateway_host::ConfigError), + #[error(transparent)] + Host(#[from] gateway_host::HostError), + #[error(transparent)] + Outbound(#[from] gateway_host::outbound::HttpError), + #[error(transparent)] + Store(#[from] gateway_store::StoreError), + #[error(transparent)] + CommandStoreDrain(#[from] gateway_store::CommandStoreDrainError), + #[error(transparent)] + OpenAi(#[from] provider_openai::OpenAiInitializeError), + #[error(transparent)] + Xai(#[from] provider_xai::XaiInitializeError), + #[error(transparent)] + Registry(#[from] gateway_core::engine::provider::RegistryError), + #[error(transparent)] + Core(#[from] gateway_core::CoreError), + #[error(transparent)] + Admin(#[from] gateway_admin::model::AdminError), + #[error(transparent)] + Api(#[from] gateway_api::ApiError), +} diff --git a/backend/apps/gateway/src/bootstrap/plugins.rs b/backend/apps/gateway/src/bootstrap/plugins.rs new file mode 100644 index 000000000..33c3ae1ac --- /dev/null +++ b/backend/apps/gateway/src/bootstrap/plugins.rs @@ -0,0 +1,108 @@ +//! 插件 Runtime 的组装与回调资源持有；服务和命令模式复用同一组管理端口。 + +use std::sync::Arc; + +use gateway_admin::{ + model::AdminError, + ports::{ + plugin_accounts::PluginAccountAccess, plugin_client_keys::PluginClientKeyAccess, + plugin_resources::PluginResourceAccess, provider::ProviderAdminRegistry, + }, +}; +use gateway_core::runtime::{RuntimeSnapshotHandle, SnapshotControl}; +use gateway_host::ConfigError; +use gateway_plugin_runtime::PluginRuntime; + +use super::BootstrapError; + +// 必须在调用方仍持有 Core 和回调资源时异步关闭，不能先退出该作用域再清理。 +macro_rules! or_shutdown { + ($runtime:expr, $result:expr) => { + match $result { + Ok(value) => value, + Err(error) => { + $runtime.shutdown().await; + return Err(error.into()); + } + } + }; +} + +pub(super) use or_shutdown; + +pub(super) struct ManagementPorts { + // Runtime 保存 Weak；组合根持有这些 Arc，直到插件全部关闭。 + pub(super) accounts: Arc, + keys: Arc, + resources: Arc, +} + +impl ManagementPorts { + pub(super) fn new( + store: &gateway_store::StoreBundle, + providers: ProviderAdminRegistry, + snapshot: Arc, + ) -> Self { + let ports = store.admin_ports(); + Self { + accounts: gateway_admin::initialize_plugin_accounts( + providers.clone(), + ports.accounts(), + snapshot.clone(), + ), + keys: gateway_admin::initialize_plugin_client_keys( + providers, + ports.client_keys(), + snapshot.clone(), + ), + resources: gateway_admin::initialize_plugin_resources( + ports.plugin_resources(), + snapshot, + ), + } + } + + pub(super) fn bind(&self, runtime: &PluginRuntime) -> Result<(), AdminError> { + runtime.bind_account_ports(&self.accounts)?; + runtime.bind_client_key_ports(&self.keys)?; + runtime.bind_resource_ports(&self.resources) + } +} + +pub(super) fn service_middleware( + snapshots: RuntimeSnapshotHandle, + runtime: &PluginRuntime, +) -> gateway_admin::service::PlanSource { + let middleware = runtime.middleware_registry(); + Arc::new(move || { + let snapshot = snapshots.snapshot_for_diagnostics()?; + middleware.resolve(snapshot.extensions()?) + }) +} + +pub(super) fn prepare_plugins( + store: &gateway_store::StoreBundle, + plugin_cache: std::path::PathBuf, + version: &str, + http: Arc, +) -> Result, BootstrapError> { + Ok(Arc::new( + gateway_plugin_runtime::PluginRuntime::new( + store.admin_ports().plugins(), + store.admin_ports().plugin_state(), + gateway_plugin_runtime::PluginRuntimeConfig { + cache_directory: plugin_cache, + host_version: version + .trim_start_matches('v') + .parse() + .map_err(|_| ConfigError::InvalidField("host.version"))?, + package_limits: gateway_plugin_runtime::PackageLimits::default(), + rpc_limits: gateway_plugin_runtime::RpcLimits::default(), + restart_circuit: Default::default(), + }, + http, + Arc::new(gateway_host::process::ProcessSupervisor::default()), + ) + .with_oauth_pending(store.provider_ports().oauth_pending()), + )) +} diff --git a/backend/apps/gateway/src/bootstrap/server.rs b/backend/apps/gateway/src/bootstrap/server.rs new file mode 100644 index 000000000..40c0c14e3 --- /dev/null +++ b/backend/apps/gateway/src/bootstrap/server.rs @@ -0,0 +1,164 @@ +//! 服务模式完成全部组装后交给 Host 监听，并在请求和 Worker 排空后关闭插件。 + +use gateway_host::ConfigError; + +use super::{ + BootstrapError, GatewayConfig, + plugins::{ManagementPorts, or_shutdown, prepare_plugins, service_middleware}, + startup::{Environment, Mode, Providers}, +}; + +/// 按冻结顺序初始化全部 Bundle，并把进程阻塞权交给 Host。 +pub async fn run() -> Result<(), BootstrapError> { + let config = gateway_host::load_config::()?; + let environment = Environment::initialize(config.host, config.store, Mode::Server).await?; + let mut providers = Providers::initialize(config.openai, &environment).await?; + let Environment { + host, + mut store, + timezone, + plugin_cache, + version, + plugin_http, + } = environment; + let plugin_runtime = prepare_plugins(&store, plugin_cache, &version, plugin_http.clone())?; + let core = gateway_core::prepare( + store.core_ports(), + providers.core, + Some(plugin_runtime.clone()), + Some(plugin_runtime.observer_registry()), + Some(plugin_runtime.policy_registry()), + Some(plugin_runtime.middleware_registry()), + Some(plugin_runtime.frontend_authentication_registry()), + ); + let management = ManagementPorts::new(&store, providers.admin.clone(), core.snapshot_control()); + or_shutdown!(plugin_runtime, management.bind(&plugin_runtime)); + let mut core = or_shutdown!(plugin_runtime, core.activate().await); + let nested_models = core.nested_model_execution_port(); + let affinity_lookup = core.affinity_lookup_port(); + or_shutdown!( + plugin_runtime, + plugin_runtime.bind_model_ports(&nested_models, &affinity_lookup) + ); + host.report_startup_ready("Core"); + let host_version = or_shutdown!( + plugin_runtime, + version + .trim_start_matches('v') + .parse() + .map_err(|_| ConfigError::InvalidField("host.version")) + ); + let plugin_inspector = std::sync::Arc::new(gateway_plugin_runtime::PackageInspector::new( + gateway_plugin_runtime::PackageLimits::default(), + host_version, + )); + let plugin_distribution = std::sync::Arc::new(or_shutdown!( + plugin_runtime, + gateway_host::plugin_distribution::HttpPluginDistribution::new(plugin_http) + )); + let mut admin = or_shutdown!( + plugin_runtime, + gateway_admin::initialize_with_plugin_accounts( + config.admin, + config.client, + store.admin_ports(), + gateway_admin::AdminRuntimePorts { + timezone, + service_middleware: service_middleware(core.snapshots(), &plugin_runtime), + providers: providers.admin, + plugin_preparation: plugin_runtime.clone(), + plugin_management: plugin_runtime.clone(), + published_snapshot: core.snapshots(), + plugin_inspector, + plugin_distribution, + pricing_source: std::sync::Arc::new(gateway_host::pricing::ModelsDevPricing), + snapshot: core.snapshot_control(), + account_probe: core.account_probe(), + proxy_probe: host.proxy_probe(provider_openai::build_reqwest_client_with_custom_ca), + client_distribution: host.client_distribution_resolver(), + system: host.system_operations(), + client_key_verifier: core.client_key_verifier(), + }, + management.accounts.clone(), + ) + .await + ); + let official_files = host.official_plugin_release_files(); + let official_identity = host.official_plugin_release_identity(); + let official_import = or_shutdown!( + plugin_runtime, + admin + .services() + .plugins() + .import_official_release( + official_files.as_ref(), + &official_identity, + &gateway_admin::model::MutationContext { + actor: gateway_admin::model::MutationActor::System, + request_id: "startup-official-plugin-import".to_owned(), + }, + ) + .await + ); + if official_import.artifacts > 0 { + host.report_startup_ready("Official plugins"); + } + host.report_startup_ready("Admin"); + + let mut probes = store.health_probes(); + probes.extend(core.health_probes()); + probes.push(host.logging_health_probe()); + let api = or_shutdown!( + plugin_runtime, + gateway_api::initialize( + config.api, + core.execution_service(), + admin.services(), + probes, + host.worker_health(), + host.connection_lifecycle(), + ) + ) + .with_middleware({ + let snapshots = core.snapshots(); + let middleware = plugin_runtime.middleware_registry(); + move |snapshot| { + if let Some(snapshot) = snapshot { + return middleware.resolve(snapshot.extensions()?); + } + // 数据面未就绪时，管理与诊断路由仍可使用发布候选中的插件。 + let diagnostic = snapshots.snapshot_for_diagnostics()?; + middleware.resolve(diagnostic.extensions()?) + } + }); + let http_dispatcher = api.dispatcher(); + or_shutdown!(plugin_runtime, plugin_runtime.bind_http(&http_dispatcher)); + host.report_startup_ready("API"); + + or_shutdown!( + plugin_runtime, + plugin_runtime.bind_services(&admin.services().public_services()) + ); + + let mut plan = store.take_worker_contributions(); + plan.push(or_shutdown!( + plugin_runtime, + gateway_host::retention::worker(store.retention()) + )); + plan.extend(core.take_worker_contributions()); + plan.extend(providers.openai.take_worker_contributions()); + plan.extend(providers.xai.take_worker_contributions()); + plan.extend(admin.take_worker_contributions()); + plan.push(or_shutdown!( + plugin_runtime, + plugin_runtime.maintenance_worker(core.snapshots()) + )); + or_shutdown!( + plugin_runtime, + host.start_workers(plan, store.worker_leader_lease()) + ); + host.report_startup_ready("Workers"); + let served = host.serve(api.router()).await; + plugin_runtime.shutdown().await; + Ok(served?) +} diff --git a/backend/apps/gateway/src/bootstrap/startup.rs b/backend/apps/gateway/src/bootstrap/startup.rs new file mode 100644 index 000000000..1ad375e49 --- /dev/null +++ b/backend/apps/gateway/src/bootstrap/startup.rs @@ -0,0 +1,94 @@ +//! 启动模式共享的基础设施与 Provider 组装，不启动 HTTP 或 Worker。 + +use std::{path::PathBuf, sync::Arc}; + +use gateway_admin::ports::provider::ProviderAdminRegistry; +use gateway_core::{engine::provider::ProviderRegistry, time::DeploymentTimeZone}; +use gateway_host::{ConfigError, HostBundle, HostConfig, outbound::HttpClient}; +use gateway_store::{StoreBundle, StoreConfig}; + +use super::BootstrapError; + +pub(super) enum Mode { + Server, + Command, + Help, +} + +pub(super) struct Environment { + pub(super) host: HostBundle, + pub(super) store: StoreBundle, + pub(super) timezone: DeploymentTimeZone, + pub(super) plugin_cache: PathBuf, + pub(super) version: String, + pub(super) plugin_http: Arc, +} + +impl Environment { + pub(super) async fn initialize( + host: HostConfig, + store: StoreConfig, + mode: Mode, + ) -> Result { + let timezone = host.timezone; + let store = store.with_timezone(timezone); + let plugin_cache = host.runtime_data_dir().join("plugins"); + let host = match mode { + Mode::Server => gateway_host::initialize(host).await?, + Mode::Command | Mode::Help => gateway_host::initialize_command_line(host).await?, + }; + host.report_startup_ready("Host"); + let store = match mode { + Mode::Server => gateway_store::initialize(store).await?, + Mode::Command => gateway_store::initialize_command_line(store).await?, + Mode::Help => gateway_store::initialize_read_only(store).await?, + }; + host.report_startup_ready("Store"); + let version = host + .system_operations() + .version() + .await + .map_err(|_| ConfigError::InvalidField("host.version"))? + .version; + let plugin_http = Arc::new(HttpClient::new()?); + Ok(Self { + host, + store, + timezone, + plugin_cache, + version, + plugin_http, + }) + } +} + +pub(super) struct Providers { + pub(super) openai: provider_openai::ProviderBundle, + pub(super) xai: provider_xai::ProviderBundle, + pub(super) core: ProviderRegistry, + pub(super) admin: ProviderAdminRegistry, +} + +impl Providers { + pub(super) async fn initialize( + config: provider_openai::OpenAiConfig, + environment: &Environment, + ) -> Result { + let ports = environment.store.provider_ports(); + let openai = + provider_openai::initialize(config.with_timezone(environment.timezone), ports.clone()) + .await?; + environment.host.report_startup_ready("OpenAI Provider"); + let xai = provider_xai::initialize(ports).await?; + environment.host.report_startup_ready("xAI Provider"); + let core = ProviderRegistry::new([openai.core_provider(), xai.core_provider()])?; + let admin = ProviderAdminRegistry::new([openai.admin_provider(), xai.admin_provider()]) + .map_err(|_| gateway_admin::model::AdminError::invalid("Provider 管理身份冲突"))?; + Ok(Self { + openai, + xai, + core, + admin, + }) + } +} diff --git a/backend/apps/gateway/tests/bootstrap/command.rs b/backend/apps/gateway/tests/bootstrap/command.rs new file mode 100644 index 000000000..a944ab5a2 --- /dev/null +++ b/backend/apps/gateway/tests/bootstrap/command.rs @@ -0,0 +1,49 @@ +use std::{fs, process::Command}; + +#[test] +fn basic_cli_options_do_not_load_configuration_and_reject_unexpected_arguments() { + let directory = tempfile::tempdir().unwrap(); + fs::create_dir(directory.path().join("deploy")).unwrap(); + fs::write(directory.path().join("deploy/config.yaml"), "invalid: [").unwrap(); + for flag in ["--help", "-h", "help", "--version", "-V"] { + let output = Command::new(env!("CARGO_BIN_EXE_codex-proxy-rs")) + .arg(flag) + .current_dir(directory.path()) + .env_clear() + .output() + .unwrap(); + assert!(output.status.success(), "{flag}"); + assert!(!output.stdout.is_empty()); + assert!(output.stderr.is_empty()); + } + for args in [ + vec!["unknown"], + vec!["--help", "extra"], + vec!["serve", "extra"], + ] { + let output = Command::new(env!("CARGO_BIN_EXE_codex-proxy-rs")) + .args(args) + .current_dir(directory.path()) + .env_clear() + .output() + .unwrap(); + assert!(!output.status.success()); + assert!(output.stdout.is_empty()); + assert!(!String::from_utf8_lossy(&output.stderr).contains("configuration")); + } +} + +#[cfg(unix)] +#[test] +fn non_utf8_command_is_not_interpreted_as_serve() { + use std::os::unix::ffi::OsStringExt as _; + + let output = Command::new(env!("CARGO_BIN_EXE_codex-proxy-rs")) + .arg(std::ffi::OsString::from_vec(vec![0xff])) + .env_clear() + .output() + .unwrap(); + assert!(!output.status.success()); + assert!(output.stdout.is_empty()); + assert!(String::from_utf8_lossy(&output.stderr).contains("command must be UTF-8")); +} diff --git a/backend/apps/gateway/tests/bootstrap.rs b/backend/apps/gateway/tests/bootstrap/config.rs similarity index 68% rename from backend/apps/gateway/tests/bootstrap.rs rename to backend/apps/gateway/tests/bootstrap/config.rs index 6fd714153..e2fb46a0a 100644 --- a/backend/apps/gateway/tests/bootstrap.rs +++ b/backend/apps/gateway/tests/bootstrap/config.rs @@ -3,205 +3,12 @@ use std::{fs, process::Command}; use codex_proxy_rs::bootstrap::GatewayConfig; use gateway_host::LoadableConfig; -const CONFIG_EXAMPLE: &str = include_str!("../../../../deploy/config.example.yaml"); +const CONFIG_EXAMPLE: &str = include_str!("../../../../../deploy/config.example.yaml"); const POSTGRES_PASSWORD: &str = "111111111111111111111111111111111111111111111111"; const REDIS_PASSWORD: &str = "222222222222222222222222222222222222222222222222"; const ADMIN_PASSWORD: &str = "test-admin-password"; const TOPOLOGY_CHILD_ENV: &str = "CPR_TEST_TOPOLOGY_CHILD"; -#[test] -fn basic_cli_options_do_not_load_configuration_and_reject_unexpected_arguments() { - let directory = tempfile::tempdir().unwrap(); - fs::create_dir(directory.path().join("deploy")).unwrap(); - fs::write(directory.path().join("deploy/config.yaml"), "invalid: [").unwrap(); - for flag in ["--help", "-h", "help", "--version", "-V"] { - let output = Command::new(env!("CARGO_BIN_EXE_codex-proxy-rs")) - .arg(flag) - .current_dir(directory.path()) - .env_clear() - .output() - .unwrap(); - assert!(output.status.success(), "{flag}"); - assert!(!output.stdout.is_empty()); - assert!(output.stderr.is_empty()); - } - for args in [ - vec!["unknown"], - vec!["--help", "extra"], - vec!["serve", "extra"], - ] { - let output = Command::new(env!("CARGO_BIN_EXE_codex-proxy-rs")) - .args(args) - .current_dir(directory.path()) - .env_clear() - .output() - .unwrap(); - assert!(!output.status.success()); - assert!(output.stdout.is_empty()); - assert!(!String::from_utf8_lossy(&output.stderr).contains("configuration")); - } -} - -#[cfg(unix)] -#[test] -fn non_utf8_command_is_not_interpreted_as_serve() { - use std::os::unix::ffi::OsStringExt as _; - - let output = Command::new(env!("CARGO_BIN_EXE_codex-proxy-rs")) - .arg(std::ffi::OsString::from_vec(vec![0xff])) - .env_clear() - .output() - .unwrap(); - assert!(!output.status.success()); - assert!(output.stdout.is_empty()); - assert!(String::from_utf8_lossy(&output.stderr).contains("command must be UTF-8")); -} - -#[tokio::test] -async fn proxy_probe_should_use_provider_custom_ca_for_https_proxies() { - use gateway_admin::ports::proxy::ProxyProbe; - use gateway_core::account::OutboundProxy; - use gateway_host::proxy_probe::HttpProxyProbe; - use std::{sync::Arc, time::Duration}; - use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _}; - use tokio_rustls::{ - TlsAcceptor, - rustls::{ - ServerConfig, - pki_types::{CertificateDer, PrivateKeyDer, pem::PemObject as _}, - }, - }; - - const CHILD_ENV: &str = "CPR_TEST_PROXY_TLS_DIRECTORY"; - let Ok(directory) = std::env::var(CHILD_ENV) else { - let directory = tempfile::tempdir().unwrap(); - generate_proxy_test_certificates(directory.path()); - // 使用子进程隔离环境变量，避免并行测试读取到临时 CA 配置。 - for ca_env in ["CODEX_CA_CERTIFICATE", "SSL_CERT_FILE"] { - let mut child = Command::new(std::env::current_exe().unwrap()); - child - .args([ - "--exact", - "bootstrap::proxy_probe_should_use_provider_custom_ca_for_https_proxies", - "--nocapture", - ]) - .env(CHILD_ENV, directory.path()) - .env_remove("CODEX_CA_CERTIFICATE") - .env_remove("SSL_CERT_FILE") - .env(ca_env, directory.path().join("ca.pem")); - if ca_env == "CODEX_CA_CERTIFICATE" { - child.env( - "SSL_CERT_FILE", - directory.path().join("missing-fallback.pem"), - ); - } - let output = child.output().unwrap(); - assert!( - output.status.success(), - "{ca_env}: {}\n{}", - String::from_utf8_lossy(&output.stdout), - String::from_utf8_lossy(&output.stderr) - ); - } - return; - }; - provider_openai::ensure_rustls_provider(); - let directory = std::path::Path::new(&directory); - let certificate = CertificateDer::from_pem_file(directory.join("server.pem")).unwrap(); - let key = PrivateKeyDer::from_pem_file(directory.join("server.key")).unwrap(); - let acceptor = TlsAcceptor::from(Arc::new( - ServerConfig::builder() - .with_no_client_auth() - .with_single_cert(vec![certificate], key) - .unwrap(), - )); - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let proxy = - OutboundProxy::parse(&format!("https://{}", listener.local_addr().unwrap())).unwrap(); - let server = tokio::spawn(async move { - let (socket, _) = listener.accept().await.unwrap(); - let mut stream = acceptor.accept(socket).await.unwrap(); - let mut request = Vec::new(); - while !request.ends_with(b"\r\n\r\n") { - assert!(request.len() < 8192); - request.push(stream.read_u8().await.unwrap()); - } - assert!(request.starts_with(b"GET http://unresolvable.invalid/ip ")); - let body = "{\"ip\":\"203.0.113.8\"}"; - stream.write_all(format!("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()).as_bytes()).await.unwrap(); - stream.shutdown().await.unwrap(); - }); - let probe = HttpProxyProbe::new("http://unresolvable.invalid/ip") - .with_client_builder(provider_openai::build_reqwest_client_with_custom_ca); - let result = probe.test(&proxy, false).await; - assert!(result.success, "{}", result.message); - assert_eq!(result.exit_ip.unwrap().to_string(), "203.0.113.8"); - tokio::time::timeout(Duration::from_secs(5), server) - .await - .unwrap() - .unwrap(); -} - -fn generate_proxy_test_certificates(directory: &std::path::Path) { - let openssl = |args: &[&str]| { - let output = Command::new("openssl") - .args(args) - .current_dir(directory) - .output() - .unwrap(); - assert!( - output.status.success(), - "{}", - String::from_utf8_lossy(&output.stderr) - ); - }; - openssl(&[ - "req", - "-x509", - "-newkey", - "rsa:2048", - "-nodes", - "-keyout", - "ca.key", - "-out", - "ca.pem", - "-days", - "1", - "-subj", - "/CN=Proxy Test CA", - ]); - openssl(&[ - "req", - "-newkey", - "rsa:2048", - "-nodes", - "-keyout", - "server.key", - "-out", - "server.csr", - "-subj", - "/CN=localhost", - ]); - fs::write(directory.join("extensions"), "subjectAltName=IP:127.0.0.1\nbasicConstraints=critical,CA:FALSE\nkeyUsage=digitalSignature,keyEncipherment\nextendedKeyUsage=serverAuth\n").unwrap(); - openssl(&[ - "x509", - "-req", - "-in", - "server.csr", - "-CA", - "ca.pem", - "-CAkey", - "ca.key", - "-CAcreateserial", - "-out", - "server.pem", - "-days", - "1", - "-extfile", - "extensions", - ]); -} - #[test] fn config_loader_should_load_complete_terminal_example() { parse_config(&valid_config()).expect("terminal config example"); @@ -230,7 +37,7 @@ fn config_loader_should_share_resolved_assets_with_system_update() { child .args([ "--exact", - "bootstrap::config_loader_should_share_resolved_assets_with_system_update", + "bootstrap::config::config_loader_should_share_resolved_assets_with_system_update", ]) .env(CHILD_ENV, case) .env_remove("CPR_WEB_DIST_DIR"); @@ -292,7 +99,7 @@ fn config_loader_should_apply_only_explicit_topology_overrides() { let status = Command::new(std::env::current_exe().expect("current test executable")) .args([ "--exact", - "bootstrap::config_loader_should_apply_only_explicit_topology_overrides", + "bootstrap::config::config_loader_should_apply_only_explicit_topology_overrides", ]) .env(TOPOLOGY_CHILD_ENV, "1") .env("CPR_SERVER_HOST", "127.0.0.1") @@ -410,7 +217,7 @@ fn config_loader_should_report_startup_configuration_diagnostics() { let output = Command::new(std::env::current_exe().unwrap()) .args([ "--exact", - "bootstrap::config_loader_should_report_startup_configuration_diagnostics", + "bootstrap::config::config_loader_should_report_startup_configuration_diagnostics", "--nocapture", ]) .env(CHILD_ENV, "1") diff --git a/backend/apps/gateway/tests/bootstrap/mod.rs b/backend/apps/gateway/tests/bootstrap/mod.rs new file mode 100644 index 000000000..98de9f4f2 --- /dev/null +++ b/backend/apps/gateway/tests/bootstrap/mod.rs @@ -0,0 +1,3 @@ +mod command; +mod config; +mod server; diff --git a/backend/apps/gateway/tests/bootstrap/server.rs b/backend/apps/gateway/tests/bootstrap/server.rs new file mode 100644 index 000000000..d933aecb4 --- /dev/null +++ b/backend/apps/gateway/tests/bootstrap/server.rs @@ -0,0 +1,146 @@ +use std::{fs, process::Command}; + +#[tokio::test] +async fn proxy_probe_should_use_provider_custom_ca_for_https_proxies() { + use gateway_admin::ports::proxy::ProxyProbe; + use gateway_core::account::OutboundProxy; + use gateway_host::proxy_probe::HttpProxyProbe; + use std::{sync::Arc, time::Duration}; + use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _}; + use tokio_rustls::{ + TlsAcceptor, + rustls::{ + ServerConfig, + pki_types::{CertificateDer, PrivateKeyDer, pem::PemObject as _}, + }, + }; + + const CHILD_ENV: &str = "CPR_TEST_PROXY_TLS_DIRECTORY"; + let Ok(directory) = std::env::var(CHILD_ENV) else { + let directory = tempfile::tempdir().unwrap(); + generate_proxy_test_certificates(directory.path()); + // 使用子进程隔离环境变量，避免并行测试读取到临时 CA 配置。 + for ca_env in ["CODEX_CA_CERTIFICATE", "SSL_CERT_FILE"] { + let mut child = Command::new(std::env::current_exe().unwrap()); + child + .args([ + "--exact", + "bootstrap::server::proxy_probe_should_use_provider_custom_ca_for_https_proxies", + "--nocapture", + ]) + .env(CHILD_ENV, directory.path()) + .env_remove("CODEX_CA_CERTIFICATE") + .env_remove("SSL_CERT_FILE") + .env(ca_env, directory.path().join("ca.pem")); + if ca_env == "CODEX_CA_CERTIFICATE" { + child.env( + "SSL_CERT_FILE", + directory.path().join("missing-fallback.pem"), + ); + } + let output = child.output().unwrap(); + assert!( + output.status.success(), + "{ca_env}: {}\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + } + return; + }; + provider_openai::ensure_rustls_provider(); + let directory = std::path::Path::new(&directory); + let certificate = CertificateDer::from_pem_file(directory.join("server.pem")).unwrap(); + let key = PrivateKeyDer::from_pem_file(directory.join("server.key")).unwrap(); + let acceptor = TlsAcceptor::from(Arc::new( + ServerConfig::builder() + .with_no_client_auth() + .with_single_cert(vec![certificate], key) + .unwrap(), + )); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let proxy = + OutboundProxy::parse(&format!("https://{}", listener.local_addr().unwrap())).unwrap(); + let server = tokio::spawn(async move { + let (socket, _) = listener.accept().await.unwrap(); + let mut stream = acceptor.accept(socket).await.unwrap(); + let mut request = Vec::new(); + while !request.ends_with(b"\r\n\r\n") { + assert!(request.len() < 8192); + request.push(stream.read_u8().await.unwrap()); + } + assert!(request.starts_with(b"GET http://unresolvable.invalid/ip ")); + let body = "{\"ip\":\"203.0.113.8\"}"; + stream.write_all(format!("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()).as_bytes()).await.unwrap(); + stream.shutdown().await.unwrap(); + }); + let probe = HttpProxyProbe::new("http://unresolvable.invalid/ip") + .with_client_builder(provider_openai::build_reqwest_client_with_custom_ca); + let result = probe.test(&proxy, false).await; + assert!(result.success, "{}", result.message); + assert_eq!(result.exit_ip.unwrap().to_string(), "203.0.113.8"); + tokio::time::timeout(Duration::from_secs(5), server) + .await + .unwrap() + .unwrap(); +} + +fn generate_proxy_test_certificates(directory: &std::path::Path) { + let openssl = |args: &[&str]| { + let output = Command::new("openssl") + .args(args) + .current_dir(directory) + .output() + .unwrap(); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + }; + openssl(&[ + "req", + "-x509", + "-newkey", + "rsa:2048", + "-nodes", + "-keyout", + "ca.key", + "-out", + "ca.pem", + "-days", + "1", + "-subj", + "/CN=Proxy Test CA", + ]); + openssl(&[ + "req", + "-newkey", + "rsa:2048", + "-nodes", + "-keyout", + "server.key", + "-out", + "server.csr", + "-subj", + "/CN=localhost", + ]); + fs::write(directory.join("extensions"), "subjectAltName=IP:127.0.0.1\nbasicConstraints=critical,CA:FALSE\nkeyUsage=digitalSignature,keyEncipherment\nextendedKeyUsage=serverAuth\n").unwrap(); + openssl(&[ + "x509", + "-req", + "-in", + "server.csr", + "-CA", + "ca.pem", + "-CAkey", + "ca.key", + "-CAcreateserial", + "-out", + "server.pem", + "-days", + "1", + "-extfile", + "extensions", + ]); +} diff --git a/backend/apps/gateway/tests/main.rs b/backend/apps/gateway/tests/main.rs index c99574358..f989ed589 100644 --- a/backend/apps/gateway/tests/main.rs +++ b/backend/apps/gateway/tests/main.rs @@ -18,7 +18,12 @@ fn app_tree_matches_frozen_terminal_manifest() { assert_eq!( rust_files(&root.join("src")), BTreeSet::from([ - PathBuf::from("bootstrap.rs"), + PathBuf::from("bootstrap/command.rs"), + PathBuf::from("bootstrap/config.rs"), + PathBuf::from("bootstrap/mod.rs"), + PathBuf::from("bootstrap/plugins.rs"), + PathBuf::from("bootstrap/server.rs"), + PathBuf::from("bootstrap/startup.rs"), PathBuf::from("lib.rs"), PathBuf::from("main.rs"), ]), @@ -27,7 +32,10 @@ fn app_tree_matches_frozen_terminal_manifest() { rust_files(&root.join("tests")), BTreeSet::from([ PathBuf::from("architecture.rs"), - PathBuf::from("bootstrap.rs"), + PathBuf::from("bootstrap/command.rs"), + PathBuf::from("bootstrap/config.rs"), + PathBuf::from("bootstrap/mod.rs"), + PathBuf::from("bootstrap/server.rs"), PathBuf::from("main.rs"), ]), );