diff --git a/rustfs/src/app/context.rs b/rustfs/src/app/context.rs index 0d5688c0b..ae32edd80 100644 --- a/rustfs/src/app/context.rs +++ b/rustfs/src/app/context.rs @@ -20,11 +20,13 @@ mod global; mod handles; mod interfaces; mod runtime_sources; +mod server_slot; mod startup; pub use global::*; pub use handles::*; pub use interfaces::*; +pub use server_slot::ServerContextSlot; use super::storage_api::context::bucket::metadata_sys::BucketMetadataSys; use super::storage_api::context::runtime::{ @@ -1216,6 +1218,7 @@ mod tests { assert_eq!(context_server_config_published.load(Ordering::SeqCst), 1); assert!(publish_storage_class_config_with(Some(context.clone()), StorageClassConfig::default())); assert_eq!(context_storage_class_published.load(Ordering::SeqCst), 1); + let slot_context = context.clone(); assert_eq!( resolve_buffer_config_with(Some(context)) .expect("context buffer config") @@ -1258,5 +1261,19 @@ mod tests { assert!(!publish_server_config_with(None, Config::new())); assert!(!publish_storage_class_config_with(None, StorageClassConfig::default())); assert!(resolve_buffer_config_with(None).is_none()); + + // backlog#1052 S2: a per-server context slot resolves its *installed* + // context — not whatever another server installed — and the first + // installation wins. + let slot = ServerContextSlot::new(); + assert!(slot.install(slot_context.clone()), "first install must succeed"); + assert!(!slot.install(slot_context.clone()), "the slot is install-once"); + let resolved = slot.app_context().expect("installed slot must resolve its context"); + assert!(Arc::ptr_eq(&resolved, &slot_context), "the slot must hand back the installed context"); + let resolved_store = slot.object_store().expect("installed slot must resolve the store"); + assert!( + Arc::ptr_eq(&resolved_store, &slot_context.object_store()), + "the slot must dispatch to the installed context's store" + ); } } diff --git a/rustfs/src/app/context/server_slot.rs b/rustfs/src/app/context/server_slot.rs new file mode 100644 index 000000000..34f6f1ee4 --- /dev/null +++ b/rustfs/src/app/context/server_slot.rs @@ -0,0 +1,88 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Per-server application-context slot (backlog#1052 S2). +//! +//! The S3 service starts serving before storage and IAM finish initializing, +//! so a server cannot hold its `AppContext` at construction time. This slot is +//! the late-bound carrier: startup creates one per server, hands it to the +//! request-path owners (the `FS` service), and installs the `AppContext` into +//! it once IAM bootstrap completes. +//! +//! Until every app subsystem is per-server (backlog#1052 S3), resolution falls +//! back to the process-global `AppContext` singleton when the slot has not +//! been installed — byte-for-byte the ambient resolution the request path used +//! before this seam existed. The fallback is removed when multi-instance flips +//! on (backlog#1052 S5). + +use super::global::{AppContext, get_global_app_context}; +use crate::app::storage_api::context::ECStore; +use std::sync::{Arc, OnceLock}; + +/// Late-bound, per-server handle to the application context. +#[derive(Default)] +pub struct ServerContextSlot { + app_context: OnceLock>, +} + +// Manual Debug: AppContext is not Debug, so summarize installation state. +impl std::fmt::Debug for ServerContextSlot { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ServerContextSlot") + .field("app_context_installed", &self.app_context.get().is_some()) + .finish() + } +} + +impl ServerContextSlot { + pub fn new() -> Arc { + Arc::new(Self { + app_context: OnceLock::new(), + }) + } + + /// Install this server's application context (once). Returns `false` if + /// the slot was already installed; the first installation wins, matching + /// the process-global singleton's `get_or_init` semantics. + pub fn install(&self, context: Arc) -> bool { + self.app_context.set(context).is_ok() + } + + /// This server's application context: the installed one, or the + /// process-global singleton as the single-instance legacy default. + pub fn app_context(&self) -> Option> { + self.app_context.get().cloned().or_else(get_global_app_context) + } + + /// This server's object store, resolved through [`Self::app_context`]. + pub fn object_store(&self) -> Option> { + self.app_context().map(|context| context.object_store()) + } +} + +#[cfg(test)] +mod tests { + use super::{ServerContextSlot, get_global_app_context}; + + // Before anything is installed — and with no global AppContext in this + // process (nextest runs each test in its own process) — resolution yields + // nothing: the same "not ready yet" answer the ambient path gives before + // IAM bootstrap completes. + #[test] + fn empty_slot_resolves_like_the_ambient_path() { + let slot = ServerContextSlot::new(); + assert_eq!(slot.app_context().is_some(), get_global_app_context().is_some()); + assert_eq!(slot.object_store().is_some(), get_global_app_context().is_some()); + } +} diff --git a/rustfs/src/app/context/startup.rs b/rustfs/src/app/context/startup.rs index f726c82be..7808e6e4d 100644 --- a/rustfs/src/app/context/startup.rs +++ b/rustfs/src/app/context/startup.rs @@ -15,6 +15,7 @@ use super::super::storage_api::context::ECStore; use super::global::{AppContext, get_global_app_context, init_global_app_context}; use super::runtime_sources; +use super::server_slot::ServerContextSlot; use rustfs_kms::KmsServiceManager; use std::io::{Error, Result}; use std::sync::Arc; @@ -30,7 +31,11 @@ impl AppContext { ensure_startup_kms_interface_with(runtime_sources::kms_service_manager, runtime_sources::init_kms_service_manager) } - pub(crate) fn ensure_startup_after_iam(store: Arc, kms_interface: Arc) -> Result<()> { + pub(crate) fn ensure_startup_after_iam( + store: Arc, + kms_interface: Arc, + server_ctx: &ServerContextSlot, + ) -> Result<()> { ensure_startup_app_context_after_iam_with( || get_global_app_context().is_some(), || { @@ -40,6 +45,12 @@ impl AppContext { Ok(()) }, )?; + // Install this server's context slot (backlog#1052 S2). Today the + // context is still the process singleton, so the slot mirrors it; + // once contexts become per-server (S3) this hands each server its own. + if let Some(context) = get_global_app_context() { + let _ = server_ctx.install(context); + } Ok(()) } } diff --git a/rustfs/src/runtime_sources.rs b/rustfs/src/runtime_sources.rs index 5a2c0ed5a..f725e238f 100644 --- a/rustfs/src/runtime_sources.rs +++ b/rustfs/src/runtime_sources.rs @@ -16,7 +16,7 @@ use crate::app::context; use std::sync::Arc; pub(crate) use context::{ - AppContext, NotifyInterface, default_notify_interface as fallback_notify_interface, + AppContext, NotifyInterface, ServerContextSlot, default_notify_interface as fallback_notify_interface, default_object_data_cache_handle as fallback_object_data_cache_handle, default_outbound_tls_runtime_interface as fallback_outbound_tls_runtime_interface, default_s3select_db_interface as fallback_s3select_db_interface, diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index cf17872ce..ba6dd4ae6 100644 --- a/rustfs/src/server/http.rs +++ b/rustfs/src/server/http.rs @@ -34,7 +34,7 @@ use crate::storage_api::server::http as storage; use crate::storage_api::server::http::request_context::{RequestContext, extract_request_id_from_headers}; use crate::storage_api::server::http::rpc::InternodeRpcService; use crate::storage_api::server::http::tonic_service::make_server; -use crate::storage_api::server::http::{TONIC_RPC_PREFIX, verify_rpc_signature}; +use crate::storage_api::server::http::{ServerContextSlot, TONIC_RPC_PREFIX, verify_rpc_signature}; use bytes::Bytes; use http::{HeaderMap, Method, Request as HttpRequest, Response}; use hyper::body::Incoming; @@ -331,7 +331,11 @@ fn trace_on_response(response: &Response, latency: Duration, s } } -pub async fn start_http_server(config: &config::Config, readiness: Arc) -> Result<(ShutdownHandle, SocketAddr)> { +pub async fn start_http_server( + config: &config::Config, + readiness: Arc, + server_ctx: Arc, +) -> Result<(ShutdownHandle, SocketAddr)> { let server_addr = parse_and_resolve_address(config.address.as_str()).map_err(Error::other)?; // The listening address and port are obtained from the parameters @@ -598,7 +602,9 @@ pub async fn start_http_server(config: &config::Config, readiness: Arc Result Result Result Result<()> { // the storage path explicitly (Phase 5 follow-up, backlog#1052); a future // multi-instance server constructs its own context here instead. let instance_ctx = bootstrap_instance_ctx(); + // This server's request-path context slot (backlog#1052 S2): handed to the + // HTTP service now, installed once IAM bootstrap completes. + let server_ctx = ServerContextSlot::new(); let StartupListenContext { readiness, @@ -120,7 +124,7 @@ async fn run(config: Config) -> Result<()> { state_manager, s3_shutdown_tx, console_shutdown_tx, - } = init_startup_http_servers(&config, readiness.clone()).await?; + } = init_startup_http_servers(&config, readiness.clone(), server_ctx.clone()).await?; let StartupStorageRuntime { store, @@ -134,6 +138,7 @@ async fn run(config: Config) -> Result<()> { ctx.clone(), readiness.clone(), state_manager.clone(), + server_ctx, ) .await?; diff --git a/rustfs/src/startup_iam.rs b/rustfs/src/startup_iam.rs index b98521dcf..473fb87be 100644 --- a/rustfs/src/startup_iam.rs +++ b/rustfs/src/startup_iam.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::runtime_sources::AppContext; +use crate::runtime_sources::{AppContext, ServerContextSlot}; use crate::server::{ServiceStateManager, publish_ready_when_runtime_ready}; use crate::storage_api::startup::iam::ECStore; use rustfs_common::{GlobalReadiness, SystemStage}; @@ -76,8 +76,9 @@ async fn finalize_iam_recovery( kms_interface: Arc, readiness: Arc, state_manager: Option>, + server_ctx: Arc, ) -> Result<()> { - AppContext::ensure_startup_after_iam(store, kms_interface)?; + AppContext::ensure_startup_after_iam(store, kms_interface, &server_ctx)?; readiness.mark_stage(SystemStage::IamReady); publish_ready_when_runtime_ready(readiness.as_ref(), state_manager.as_deref()).await @@ -90,6 +91,7 @@ fn compute_backoff_interval(attempt: u64, initial: Duration, max: Duration) -> D if backoff > max { max } else { backoff } } +#[allow(clippy::too_many_arguments)] fn spawn_iam_recovery_task( initial_interval: Duration, max_interval: Duration, @@ -98,12 +100,14 @@ fn spawn_iam_recovery_task( kms_interface: Arc, readiness: Arc, state_manager: Option>, + server_ctx: Arc, ) { let init_store = store.clone(); let finalize_store = store; let finalize_kms_interface = kms_interface; let finalize_readiness = readiness; let finalize_state_manager = state_manager; + let finalize_server_ctx = server_ctx; tokio::spawn(async move { run_iam_recovery_loop( initial_interval, @@ -118,7 +122,8 @@ fn spawn_iam_recovery_task( let kms_interface = finalize_kms_interface.clone(); let readiness = finalize_readiness.clone(); let state_manager = finalize_state_manager.clone(); - Box::pin(async move { finalize_iam_recovery(store, kms_interface, readiness, state_manager).await }) + let server_ctx = finalize_server_ctx.clone(); + Box::pin(async move { finalize_iam_recovery(store, kms_interface, readiness, state_manager, server_ctx).await }) }, ) .await; @@ -333,10 +338,11 @@ pub(crate) async fn bootstrap_or_defer_iam_init( readiness: Arc, state_manager: Option>, shutdown_token: Option, + server_ctx: Arc, ) -> Result { match attempt_init_iam_sys(store.clone()).await { Ok(()) => { - AppContext::ensure_startup_after_iam(store, kms_interface)?; + AppContext::ensure_startup_after_iam(store, kms_interface, &server_ctx)?; readiness.mark_stage(SystemStage::IamReady); return Ok(IamBootstrapDisposition::ReadyInline); } @@ -362,6 +368,7 @@ pub(crate) async fn bootstrap_or_defer_iam_init( kms_interface, readiness, state_manager, + server_ctx, ); } } @@ -374,17 +381,19 @@ pub(crate) async fn bootstrap_or_defer_iam_init_with_startup_kms( readiness: Arc, state_manager: Option>, shutdown_token: Option, + server_ctx: Arc, ) -> Result { let kms_interface = AppContext::ensure_startup_kms_interface(); - bootstrap_or_defer_iam_init(store, kms_interface, readiness, state_manager, shutdown_token).await + bootstrap_or_defer_iam_init(store, kms_interface, readiness, state_manager, shutdown_token, server_ctx).await } pub(crate) async fn init_embedded_iam_runtime( store: Arc, ctx: tokio_util::sync::CancellationToken, readiness: Arc, + server_ctx: Arc, ) -> Result { - bootstrap_or_defer_iam_init_with_startup_kms(store, readiness, None, Some(ctx)).await + bootstrap_or_defer_iam_init_with_startup_kms(store, readiness, None, Some(ctx), server_ctx).await } pub(crate) async fn init_iam_runtime( @@ -392,8 +401,9 @@ pub(crate) async fn init_iam_runtime( ctx: tokio_util::sync::CancellationToken, readiness: Arc, state_manager: Arc, + server_ctx: Arc, ) -> Result { - bootstrap_or_defer_iam_init_with_startup_kms(store, readiness, Some(state_manager), Some(ctx)).await + bootstrap_or_defer_iam_init_with_startup_kms(store, readiness, Some(state_manager), Some(ctx), server_ctx).await } #[cfg(test)] diff --git a/rustfs/src/startup_server.rs b/rustfs/src/startup_server.rs index 4fe60fbac..2fd2958f1 100644 --- a/rustfs/src/startup_server.rs +++ b/rustfs/src/startup_server.rs @@ -17,6 +17,7 @@ use crate::{ config::Config, server::{ServiceState, ServiceStateManager, ShutdownHandle, start_http_server}, startup_runtime_sources, + storage_api::server::http::ServerContextSlot, storage_api::startup::runtime_sources::InstanceContext, }; use rustfs_common::GlobalReadiness; @@ -231,9 +232,13 @@ pub(crate) async fn init_embedded_startup_listen_context( }) } -pub(crate) async fn start_embedded_http_server(config: &Config, readiness: Arc) -> Result { +pub(crate) async fn start_embedded_http_server( + config: &Config, + readiness: Arc, + server_ctx: Arc, +) -> Result { let s3_config = s3_http_server_config(config); - let (shutdown_handle, bound_addr) = start_http_server(&s3_config, readiness).await?; + let (shutdown_handle, bound_addr) = start_http_server(&s3_config, readiness, server_ctx).await?; Ok(EmbeddedHttpServer { shutdown_handle, @@ -241,16 +246,22 @@ pub(crate) async fn start_embedded_http_server(config: &Config, readiness: Arc) -> Result { +pub(crate) async fn init_startup_http_servers( + config: &Config, + readiness: Arc, + server_ctx: Arc, +) -> Result { init_capacity_management().await; let state_manager = Arc::new(ServiceStateManager::new()); state_manager.update(ServiceState::Starting); let s3_config = s3_http_server_config(config); - let (s3_shutdown_tx, _) = start_http_server(&s3_config, readiness.clone()).await?; + let (s3_shutdown_tx, _) = start_http_server(&s3_config, readiness.clone(), server_ctx.clone()).await?; let console_shutdown_tx = match console_http_server_config(config) { - Some(console_config) => Some(start_http_server(&console_config, readiness).await?.0), + // The console shares the S3 server's context slot: it is the same + // logical server on a second listener. + Some(console_config) => Some(start_http_server(&console_config, readiness, server_ctx).await?.0), None => None, }; diff --git a/rustfs/src/startup_services.rs b/rustfs/src/startup_services.rs index 34ae92c3c..73988839b 100644 --- a/rustfs/src/startup_services.rs +++ b/rustfs/src/startup_services.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::storage_api::startup::services::{ECStore, EndpointServerPools}; +use crate::storage_api::startup::services::{ECStore, EndpointServerPools, ServerContextSlot}; use crate::{ config::Config, init::{init_buffer_profile_system, init_kms_system}, @@ -48,10 +48,11 @@ pub(crate) async fn init_embedded_startup_runtime_services( store: Arc, ctx: CancellationToken, readiness: Arc, + server_ctx: Arc, ) -> Result { init_embedded_optional_service_runtime(config).await; let buckets = init_embedded_bucket_metadata_runtime(store.clone()).await?; - let iam_bootstrap = init_embedded_iam_runtime(store, ctx, readiness) + let iam_bootstrap = init_embedded_iam_runtime(store, ctx, readiness, server_ctx) .await .map_err(|err| std::io::Error::other(format!("IAM bootstrap setup: {err}")))?; init_embedded_notification_runtime(endpoint_pools, buckets).await; @@ -66,6 +67,7 @@ pub(crate) async fn init_startup_runtime_services( ctx: CancellationToken, readiness: Arc, state_manager: Arc, + server_ctx: Arc, ) -> Result { init_kms_system(config).await?; @@ -76,7 +78,7 @@ pub(crate) async fn init_startup_runtime_services( init_deadlock_detector_runtime(); let buckets = init_bucket_metadata_runtime(store.clone(), ctx.clone()).await?; - let iam_bootstrap = init_iam_runtime(store.clone(), ctx.clone(), readiness, state_manager).await?; + let iam_bootstrap = init_iam_runtime(store.clone(), ctx.clone(), readiness, state_manager, server_ctx).await?; init_auth_integrations().await?; init_notification_runtime(endpoint_pools, buckets).await?; let enable_scanner = init_background_service_runtime(store.clone()).await?; diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index 70f10b4c2..a795bf8e6 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -60,7 +60,13 @@ use crate::storage::storage_api::ecfs_consumer::StorageObjectOptions as ObjectOp #[derive(Debug, Clone)] pub struct FS { - // pub store: ECStore, + /// This server's late-bound application-context slot (backlog#1052 S2). + /// + /// Handlers resolve the object store through this slot instead of the + /// ambient process singleton, so each server dispatches requests to its + /// own store once contexts become per-server. An uninstalled slot falls + /// back to the global `AppContext` — the single-instance legacy default. + server_ctx: std::sync::Arc, } #[derive(Debug, Default, serde::Deserialize)] @@ -77,9 +83,16 @@ impl Default for FS { impl FS { pub fn new() -> Self { + Self::with_server_ctx(runtime_sources::ServerContextSlot::new()) + } + + /// Build the service bound to an explicit per-server context slot + /// (backlog#1052 S2). [`FS::new`] hands out a fresh, never-installed slot, + /// which resolves through the global fallback — the legacy behavior. + pub(crate) fn with_server_ctx(server_ctx: std::sync::Arc) -> Self { rustfs_io_metrics::init_s3_metrics(); rustfs_io_metrics::init_list_objects_metrics(); - Self {} + Self { server_ctx } } async fn replication_tagging_enabled(bucket: &str, object: &str) -> bool { @@ -102,7 +115,7 @@ impl FS { object: &str, version_id: Option<&str>, ) -> S3Result>> { - let Some(store) = runtime_sources::current_object_store_handle() else { + let Some(store) = self.server_ctx.object_store() else { return Ok(std::collections::HashMap::new()); }; let opts = ObjectOptions { @@ -348,7 +361,7 @@ impl S3 for FS { &self, req: S3Request, ) -> S3Result> { - let Some(store) = runtime_sources::current_object_store_handle() else { + let Some(store) = self.server_ctx.object_store() else { return Err(s3_error!(InternalError, "Not init")); }; @@ -398,7 +411,7 @@ impl S3 for FS { validate_table_catalog_object_mutation(&bucket, &object).await?; - let Some(store) = runtime_sources::current_object_store_handle() else { + let Some(store) = self.server_ctx.object_store() else { error!( component = LOG_COMPONENT_STORAGE, subsystem = LOG_SUBSYSTEM_TAGGING, @@ -480,7 +493,7 @@ impl S3 for FS { record_s3_op(S3Operation::GetBucketAcl); let GetBucketAclInput { bucket, .. } = req.input; - let Some(store) = runtime_sources::current_object_store_handle() else { + let Some(store) = self.server_ctx.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -496,7 +509,7 @@ impl S3 for FS { &self, req: S3Request, ) -> S3Result> { - let Some(store) = runtime_sources::current_object_store_handle() else { + let Some(store) = self.server_ctx.object_store() else { return Err(s3_error!(InternalError, "Not init")); }; @@ -586,7 +599,7 @@ impl S3 for FS { &self, req: S3Request, ) -> S3Result> { - let Some(store) = runtime_sources::current_object_store_handle() else { + let Some(store) = self.server_ctx.object_store() else { return Err(s3_error!(InternalError, "Not init")); }; @@ -634,7 +647,7 @@ impl S3 for FS { } async fn get_bucket_website(&self, req: S3Request) -> S3Result> { - let Some(store) = runtime_sources::current_object_store_handle() else { + let Some(store) = self.server_ctx.object_store() else { return Err(s3_error!(InternalError, "Not init")); }; @@ -673,7 +686,7 @@ impl S3 for FS { bucket, key, version_id, .. } = req.input; - let Some(store) = runtime_sources::current_object_store_handle() else { + let Some(store) = self.server_ctx.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -703,7 +716,7 @@ impl S3 for FS { bucket, key, version_id, .. } = req.input.clone(); - let Some(store) = runtime_sources::current_object_store_handle() else { + let Some(store) = self.server_ctx.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -764,7 +777,7 @@ impl S3 for FS { record_s3_op(S3Operation::GetObjectLockConfiguration); let GetObjectLockConfigurationInput { bucket, .. } = req.input; - let Some(store) = runtime_sources::current_object_store_handle() else { + let Some(store) = self.server_ctx.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -812,7 +825,7 @@ impl S3 for FS { bucket, key, version_id, .. } = req.input.clone(); - let Some(store) = runtime_sources::current_object_store_handle() else { + let Some(store) = self.server_ctx.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -864,7 +877,7 @@ impl S3 for FS { let bucket = req.input.bucket.as_str(); let object = req.input.key.as_str(); - let Some(store) = runtime_sources::current_object_store_handle() else { + let Some(store) = self.server_ctx.object_store() else { error!( component = LOG_COMPONENT_STORAGE, subsystem = LOG_SUBSYSTEM_TAGGING, @@ -1008,7 +1021,7 @@ impl S3 for FS { } = req.input; record_s3_op(S3Operation::PutBucketAcl); - let Some(store) = runtime_sources::current_object_store_handle() else { + let Some(store) = self.server_ctx.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -1031,7 +1044,7 @@ impl S3 for FS { &self, req: S3Request, ) -> S3Result> { - let Some(store) = runtime_sources::current_object_store_handle() else { + let Some(store) = self.server_ctx.object_store() else { return Err(s3_error!(InternalError, "Not init")); }; store @@ -1056,7 +1069,7 @@ impl S3 for FS { async fn get_bucket_logging(&self, req: S3Request) -> S3Result> { record_s3_op(S3Operation::GetBucketLogging); - let Some(store) = runtime_sources::current_object_store_handle() else { + let Some(store) = self.server_ctx.object_store() else { return Err(s3_error!(InternalError, "Not init")); }; store @@ -1075,7 +1088,7 @@ impl S3 for FS { async fn put_bucket_logging(&self, req: S3Request) -> S3Result> { record_s3_op(S3Operation::PutBucketLogging); - let Some(store) = runtime_sources::current_object_store_handle() else { + let Some(store) = self.server_ctx.object_store() else { return Err(s3_error!(InternalError, "Not init")); }; store @@ -1134,7 +1147,7 @@ impl S3 for FS { &self, req: S3Request, ) -> S3Result> { - let Some(store) = runtime_sources::current_object_store_handle() else { + let Some(store) = self.server_ctx.object_store() else { return Err(s3_error!(InternalError, "Not init")); }; store @@ -1176,7 +1189,7 @@ impl S3 for FS { } async fn put_bucket_website(&self, req: S3Request) -> S3Result> { - let Some(store) = runtime_sources::current_object_store_handle() else { + let Some(store) = self.server_ctx.object_store() else { return Err(s3_error!(InternalError, "Not init")); }; store @@ -1207,7 +1220,7 @@ impl S3 for FS { let key = &req.input.key; let version_id = req.input.version_id.clone(); - let Some(store) = runtime_sources::current_object_store_handle() else { + let Some(store) = self.server_ctx.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -1249,7 +1262,7 @@ impl S3 for FS { validate_table_catalog_object_mutation(&bucket, &key).await?; - let Some(store) = runtime_sources::current_object_store_handle() else { + let Some(store) = self.server_ctx.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -1303,7 +1316,7 @@ impl S3 for FS { let Some(input_cfg) = object_lock_configuration else { return Err(s3_error!(InvalidArgument)) }; - let Some(store) = runtime_sources::current_object_store_handle() else { + let Some(store) = self.server_ctx.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -1398,7 +1411,7 @@ impl S3 for FS { validate_table_catalog_object_mutation(&bucket, &key).await?; - let Some(store) = runtime_sources::current_object_store_handle() else { + let Some(store) = self.server_ctx.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -1477,7 +1490,7 @@ impl S3 for FS { crate::storage::s3_api::tagging::validate_object_tag_set(&tagging.tag_set)?; - let Some(store) = runtime_sources::current_object_store_handle() else { + let Some(store) = self.server_ctx.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; diff --git a/rustfs/src/storage/runtime_sources.rs b/rustfs/src/storage/runtime_sources.rs index 3abe96d7a..1d73cebed 100644 --- a/rustfs/src/storage/runtime_sources.rs +++ b/rustfs/src/storage/runtime_sources.rs @@ -22,7 +22,7 @@ use rustfs_kms::ObjectEncryptionService; use rustfs_lock::LockClient; use std::sync::Arc; -pub(crate) use crate::runtime_sources::AppContext; +pub(crate) use crate::runtime_sources::{AppContext, ServerContextSlot}; pub(crate) fn current_app_context() -> Option> { root_runtime_sources::current_app_context() diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 659ab6fa1..3cd5dd530 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -518,6 +518,7 @@ pub(crate) type FileWriter = ecstore_disk::FileWriter; pub(crate) type FS = super::ecfs::FS; pub(crate) type HashReader = ecstore_rio::HashReader; pub(crate) type InstanceContext = ecstore_runtime::InstanceContext; +pub(crate) type ServerContextSlot = crate::storage::runtime_sources::ServerContextSlot; pub(crate) type LocalPeerS3Client = ecstore_rpc::LocalPeerS3Client; pub(crate) type MetricType = ecstore_metrics::MetricType; pub(crate) type ObjectPartInfo = rustfs_filemeta::ObjectPartInfo; diff --git a/rustfs/src/storage_api.rs b/rustfs/src/storage_api.rs index 19bb24bc2..ff6f9eaca 100644 --- a/rustfs/src/storage_api.rs +++ b/rustfs/src/storage_api.rs @@ -93,7 +93,7 @@ pub(crate) mod server { } pub(crate) mod http { - pub(crate) use crate::storage::storage_api::{TONIC_RPC_PREFIX, verify_rpc_signature}; + pub(crate) use crate::storage::storage_api::{ServerContextSlot, TONIC_RPC_PREFIX, verify_rpc_signature}; pub(crate) mod ecfs { pub(crate) type FS = crate::storage::storage_api::FS; @@ -215,7 +215,7 @@ pub(crate) mod startup { } pub(crate) mod services { - pub(crate) use crate::storage::storage_api::{ECStore, EndpointServerPools}; + pub(crate) use crate::storage::storage_api::{ECStore, EndpointServerPools, ServerContextSlot}; } pub(crate) mod shutdown {