mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-31 10:32:24 +00:00
fix(admin): fail closed for empty server context slots (#5452)
* fix(admin): fail closed for empty server context slots * test(embedded): cover uninitialized context slot window * test(embedded): authenticate startup probe as server B * fix(admin): satisfy strict context-slot test lints
This commit is contained in:
Zhengchao An
@@ -2655,6 +2655,10 @@ fn is_public_health_path(path: &str) -> bool {
|
||||
path == HEALTH_PREFIX || path == HEALTH_READY_PATH
|
||||
}
|
||||
|
||||
fn server_context_not_ready_error() -> S3Error {
|
||||
s3_error!(ServiceUnavailable, "server context is not ready")
|
||||
}
|
||||
|
||||
fn is_object_zip_download_token_path(method: &Method, uri: &Uri) -> bool {
|
||||
if method != Method::GET {
|
||||
return false;
|
||||
@@ -2783,6 +2787,9 @@ where
|
||||
async fn check_access(&self, req: &mut S3Request<Body>) -> S3Result<()> {
|
||||
if let Some(server_ctx) = &self.server_ctx {
|
||||
req.extensions.insert(server_ctx.clone());
|
||||
if !is_public_health_path(req.uri.path()) && server_ctx.installed_app_context().is_none() {
|
||||
return Err(server_context_not_ready_error());
|
||||
}
|
||||
}
|
||||
if parse_replication_extension_request(&req.method, &req.uri).is_some()
|
||||
|| parse_misc_extension_request(&req.method, &req.uri).is_some()
|
||||
@@ -2836,6 +2843,9 @@ where
|
||||
async fn call(&self, mut req: S3Request<Body>) -> S3Result<S3Response<Body>> {
|
||||
if let Some(server_ctx) = &self.server_ctx {
|
||||
req.extensions.insert(server_ctx.clone());
|
||||
if !is_public_health_path(req.uri.path()) && server_ctx.installed_app_context().is_none() {
|
||||
return Err(server_context_not_ready_error());
|
||||
}
|
||||
}
|
||||
if let Some(ext_req) = parse_replication_extension_request(&req.method, &req.uri) {
|
||||
return handle_replication_extension_request(&mut req, &ext_req).await;
|
||||
|
||||
@@ -76,19 +76,29 @@ pub(crate) fn object_store_from_req<B>(req: &s3s::S3Request<B>) -> Option<Arc<EC
|
||||
}
|
||||
|
||||
pub(crate) fn app_context_from_req<B>(req: &s3s::S3Request<B>) -> Option<Arc<AppContext>> {
|
||||
req.extensions
|
||||
.get::<Arc<ServerContextSlot>>()
|
||||
.and_then(|slot| slot.app_context())
|
||||
.or_else(current_app_context)
|
||||
app_context_from_extensions(&req.extensions)
|
||||
}
|
||||
|
||||
/// Resolve an application context from request extensions.
|
||||
///
|
||||
/// An injected slot identifies the server that owns the request. If that
|
||||
/// server has not finished startup yet, returning its ambient global context
|
||||
/// could select a different server, so this fails closed. Only requests with
|
||||
/// no slot retain the legacy ambient fallback.
|
||||
pub(crate) fn app_context_from_extensions(extensions: &http::Extensions) -> Option<Arc<AppContext>> {
|
||||
match extensions.get::<Arc<ServerContextSlot>>() {
|
||||
Some(slot) => slot.installed_app_context(),
|
||||
None => current_app_context(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Field-borrow form of [`object_store_from_req`] for handlers that have
|
||||
/// already moved other request fields (body, credentials) out of the request.
|
||||
pub(crate) fn object_store_from_extensions(extensions: &http::Extensions) -> Option<Arc<ECStore>> {
|
||||
extensions
|
||||
.get::<Arc<ServerContextSlot>>()
|
||||
.and_then(|slot| slot.object_store())
|
||||
.or_else(current_object_store_handle)
|
||||
match extensions.get::<Arc<ServerContextSlot>>() {
|
||||
Some(slot) => slot.installed_object_store(),
|
||||
None => current_object_store_handle(),
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn current_notification_system() -> Option<Arc<NotificationSys>> {
|
||||
@@ -168,3 +178,151 @@ pub(crate) fn publish_storage_class_config(config: StorageClassConfig) {
|
||||
root_runtime_sources::fallback_storage_class_interface().set(config);
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{
|
||||
AppContext, IamInterface, KmsInterface, ServerContextSlot, app_context_from_extensions, app_context_from_req,
|
||||
current_app_context, object_store_from_extensions, publish_test_app_context,
|
||||
};
|
||||
use crate::admin::router::{Operation, S3Router};
|
||||
use hyper::{Method, StatusCode};
|
||||
use matchit::Params;
|
||||
use rustfs_iam::{store::object::ObjectStore, sys::IamSys};
|
||||
use rustfs_kms::KmsServiceManager;
|
||||
use s3s::route::S3Route;
|
||||
use s3s::{Body, S3ErrorCode, S3Request, S3Response, s3_error};
|
||||
use std::sync::Arc;
|
||||
|
||||
struct UnreadyIam;
|
||||
|
||||
impl IamInterface for UnreadyIam {
|
||||
fn handle(&self) -> Arc<IamSys<ObjectStore>> {
|
||||
panic!("test context does not resolve IAM")
|
||||
}
|
||||
|
||||
fn is_ready(&self) -> bool {
|
||||
false
|
||||
}
|
||||
}
|
||||
|
||||
struct TestKms;
|
||||
|
||||
impl KmsInterface for TestKms {
|
||||
fn handle(&self) -> Arc<KmsServiceManager> {
|
||||
Arc::new(KmsServiceManager::new())
|
||||
}
|
||||
}
|
||||
|
||||
async fn ambient_context() -> Arc<AppContext> {
|
||||
if let Some(context) = current_app_context() {
|
||||
return context;
|
||||
}
|
||||
|
||||
let env = rustfs_test_utils::TestECStoreEnv::builder()
|
||||
.prefix("server_context_slot")
|
||||
.disk_count(1)
|
||||
.init_bucket_metadata(false)
|
||||
.build()
|
||||
.await;
|
||||
let context = Arc::new(AppContext::new(env.ecstore, Arc::new(UnreadyIam), Arc::new(TestKms)));
|
||||
publish_test_app_context(context);
|
||||
current_app_context().expect("test context must be globally published")
|
||||
}
|
||||
|
||||
fn distinct_context(context: &AppContext) -> Arc<AppContext> {
|
||||
Arc::new(AppContext::new(context.object_store(), Arc::new(UnreadyIam), Arc::new(TestKms)))
|
||||
}
|
||||
|
||||
fn request(extensions: http::Extensions) -> S3Request<Body> {
|
||||
S3Request {
|
||||
input: Body::empty(),
|
||||
method: Method::GET,
|
||||
uri: "/context-probe".parse().expect("test URI"),
|
||||
headers: http::HeaderMap::new(),
|
||||
extensions,
|
||||
credentials: None,
|
||||
region: None,
|
||||
service: None,
|
||||
trailing_headers: None,
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial_test::serial]
|
||||
async fn request_with_empty_slot_never_uses_ambient_context() {
|
||||
let ambient = ambient_context().await;
|
||||
let slot = ServerContextSlot::new();
|
||||
let mut extensions = http::Extensions::new();
|
||||
extensions.insert(slot);
|
||||
|
||||
assert!(app_context_from_extensions(&extensions).is_none());
|
||||
assert!(object_store_from_extensions(&extensions).is_none());
|
||||
assert!(app_context_from_req(&request(extensions)).is_none());
|
||||
assert!(Arc::ptr_eq(
|
||||
¤t_app_context().expect("ambient context must remain available"),
|
||||
&ambient
|
||||
));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial_test::serial]
|
||||
async fn request_with_installed_slot_resolves_only_its_context() {
|
||||
let ambient = ambient_context().await;
|
||||
let installed = distinct_context(&ambient);
|
||||
let slot = ServerContextSlot::new();
|
||||
assert!(slot.install(installed.clone()));
|
||||
let mut extensions = http::Extensions::new();
|
||||
extensions.insert(slot);
|
||||
|
||||
let resolved = app_context_from_extensions(&extensions).expect("installed slot must resolve");
|
||||
assert!(Arc::ptr_eq(&resolved, &installed));
|
||||
assert!(!Arc::ptr_eq(&resolved, &ambient));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial_test::serial]
|
||||
async fn request_without_slot_keeps_legacy_ambient_resolution() {
|
||||
let ambient = ambient_context().await;
|
||||
let extensions = http::Extensions::new();
|
||||
|
||||
let resolved = app_context_from_extensions(&extensions).expect("legacy request must use ambient context");
|
||||
assert!(Arc::ptr_eq(&resolved, &ambient));
|
||||
assert!(object_store_from_extensions(&extensions).is_some());
|
||||
}
|
||||
|
||||
struct ContextDependentAdminRoute;
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl Operation for ContextDependentAdminRoute {
|
||||
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> s3s::S3Result<S3Response<(StatusCode, Body)>> {
|
||||
app_context_from_req(&req).ok_or_else(|| s3_error!(ServiceUnavailable, "server context is not ready"))?;
|
||||
Ok(S3Response::new((StatusCode::NO_CONTENT, Body::empty())))
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial_test::serial]
|
||||
async fn context_dependent_admin_route_fails_closed_until_slot_installation() {
|
||||
let ambient = ambient_context().await;
|
||||
let slot = ServerContextSlot::new();
|
||||
let mut router = S3Router::new(false);
|
||||
router.set_server_ctx(slot.clone());
|
||||
router
|
||||
.insert(Method::GET, "/context-probe", ContextDependentAdminRoute)
|
||||
.expect("register test route");
|
||||
|
||||
let err = router
|
||||
.call(request(http::Extensions::new()))
|
||||
.await
|
||||
.expect_err("empty slot must fail closed");
|
||||
assert_eq!(err.code(), &S3ErrorCode::ServiceUnavailable);
|
||||
|
||||
assert!(slot.install(distinct_context(&ambient)));
|
||||
let response = router
|
||||
.call(request(http::Extensions::new()))
|
||||
.await
|
||||
.expect("installed slot must serve request");
|
||||
assert_eq!(response.status, Some(StatusCode::NO_CONTENT));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -20,11 +20,10 @@
|
||||
//! 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).
|
||||
//! Some legacy, non-request call sites still use the ambient compatibility
|
||||
//! resolver. Request dispatch must instead use the strict installed-context
|
||||
//! accessors: once a request carries a server slot, an empty slot means that
|
||||
//! server is not ready rather than that another server's global context applies.
|
||||
|
||||
use super::global::{AppContext, get_global_app_context};
|
||||
use crate::app::storage_api::context::ECStore;
|
||||
@@ -62,10 +61,26 @@ impl ServerContextSlot {
|
||||
self.app_context.set(context).is_ok()
|
||||
}
|
||||
|
||||
/// This server's installed application context, if startup has completed.
|
||||
///
|
||||
/// Request-scoped resolution must use this accessor so an empty slot never
|
||||
/// resolves through another server's ambient context.
|
||||
pub fn installed_app_context(&self) -> Option<Arc<AppContext>> {
|
||||
self.app_context.get().cloned()
|
||||
}
|
||||
|
||||
/// This server's installed object store, if startup has completed.
|
||||
pub fn installed_object_store(&self) -> Option<Arc<ECStore>> {
|
||||
self.installed_app_context().map(|context| context.object_store())
|
||||
}
|
||||
|
||||
/// This server's application context: the installed one, or the
|
||||
/// process-global singleton as the single-instance legacy default.
|
||||
/// process-global singleton as the legacy compatibility default.
|
||||
///
|
||||
/// This accessor is for callers without request-slot semantics. Request
|
||||
/// dispatch must use [`Self::installed_app_context`] instead.
|
||||
pub fn app_context(&self) -> Option<Arc<AppContext>> {
|
||||
self.app_context.get().cloned().or_else(get_global_app_context)
|
||||
self.installed_app_context().or_else(get_global_app_context)
|
||||
}
|
||||
|
||||
/// This server's object store, resolved through [`Self::app_context`].
|
||||
@@ -88,7 +103,7 @@ mod tests {
|
||||
// nothing: the same "not ready yet" answer the ambient path gives before
|
||||
// IAM bootstrap completes.
|
||||
#[test]
|
||||
fn empty_slot_resolves_like_the_ambient_path() {
|
||||
fn empty_slot_retains_ambient_compatibility() {
|
||||
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());
|
||||
|
||||
@@ -60,6 +60,94 @@ use std::net::SocketAddr;
|
||||
use std::path::PathBuf;
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
#[cfg(feature = "e2e-test-hooks")]
|
||||
use std::sync::{Mutex, OnceLock};
|
||||
#[cfg(feature = "e2e-test-hooks")]
|
||||
use tokio::sync::oneshot;
|
||||
|
||||
#[cfg(feature = "e2e-test-hooks")]
|
||||
struct EmbeddedStartupHook {
|
||||
port: u16,
|
||||
http_bound: oneshot::Sender<()>,
|
||||
release: oneshot::Receiver<()>,
|
||||
}
|
||||
|
||||
#[cfg(feature = "e2e-test-hooks")]
|
||||
static EMBEDDED_STARTUP_HOOK: OnceLock<Mutex<Option<EmbeddedStartupHook>>> = OnceLock::new();
|
||||
|
||||
/// Test-only gate for the point after an embedded HTTP listener binds and
|
||||
/// before its application context is installed.
|
||||
#[cfg(feature = "e2e-test-hooks")]
|
||||
#[doc(hidden)]
|
||||
pub struct EmbeddedStartupBarrier {
|
||||
http_bound: oneshot::Receiver<()>,
|
||||
release: Option<oneshot::Sender<()>>,
|
||||
}
|
||||
|
||||
#[cfg(feature = "e2e-test-hooks")]
|
||||
impl EmbeddedStartupBarrier {
|
||||
/// Wait until the next embedded server has bound its HTTP listener.
|
||||
pub async fn wait_until_http_bound(&mut self) {
|
||||
(&mut self.http_bound)
|
||||
.await
|
||||
.expect("embedded startup hook must signal after binding HTTP");
|
||||
}
|
||||
|
||||
/// Allow the paused embedded startup to install its application context.
|
||||
pub fn release(mut self) {
|
||||
if let Some(release) = self.release.take() {
|
||||
let _ = release.send(());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "e2e-test-hooks")]
|
||||
impl Drop for EmbeddedStartupBarrier {
|
||||
fn drop(&mut self) {
|
||||
if let Some(release) = self.release.take() {
|
||||
let _ = release.send(());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Pause the embedded startup for `port` after its HTTP listener binds.
|
||||
#[cfg(feature = "e2e-test-hooks")]
|
||||
#[doc(hidden)]
|
||||
pub fn pause_embedded_startup_after_http_bind(port: u16) -> EmbeddedStartupBarrier {
|
||||
let (http_bound_tx, http_bound) = oneshot::channel();
|
||||
let (release, release_rx) = oneshot::channel();
|
||||
let hook = EmbeddedStartupHook {
|
||||
port,
|
||||
http_bound: http_bound_tx,
|
||||
release: release_rx,
|
||||
};
|
||||
let mut active_hook = EMBEDDED_STARTUP_HOOK
|
||||
.get_or_init(|| Mutex::new(None))
|
||||
.lock()
|
||||
.expect("embedded startup hook lock must not be poisoned");
|
||||
assert!(active_hook.replace(hook).is_none(), "only one embedded startup hook may be active");
|
||||
|
||||
EmbeddedStartupBarrier {
|
||||
http_bound,
|
||||
release: Some(release),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "e2e-test-hooks")]
|
||||
pub(crate) async fn wait_for_embedded_startup_hook(port: u16) {
|
||||
let hook = {
|
||||
let mut active_hook = EMBEDDED_STARTUP_HOOK
|
||||
.get_or_init(|| Mutex::new(None))
|
||||
.lock()
|
||||
.expect("embedded startup hook lock must not be poisoned");
|
||||
active_hook.take_if(|hook| hook.port == port)
|
||||
};
|
||||
if let Some(hook) = hook {
|
||||
let _ = hook.http_bound.send(());
|
||||
let _ = hook.release.await;
|
||||
}
|
||||
}
|
||||
|
||||
/// Error type for embedded server operations.
|
||||
#[derive(Debug)]
|
||||
pub enum ServerError {
|
||||
|
||||
@@ -156,6 +156,9 @@ pub(crate) async fn run_embedded_startup(args: EmbeddedStartupArgs) -> Result<Em
|
||||
let bound_addr = http_server.bound_addr;
|
||||
let cancel_token = CancellationToken::new();
|
||||
|
||||
#[cfg(feature = "e2e-test-hooks")]
|
||||
crate::embedded::wait_for_embedded_startup_hook(bound_addr.port()).await;
|
||||
|
||||
let storage_runtime = match init_embedded_startup_storage_runtime(
|
||||
listen_context.server_addr,
|
||||
&endpoint_pools,
|
||||
|
||||
@@ -19,7 +19,22 @@
|
||||
use aws_sdk_s3::config::{Credentials, Region};
|
||||
use aws_sdk_s3::primitives::ByteStream;
|
||||
use aws_sdk_s3::{Client, Config};
|
||||
#[cfg(feature = "e2e-test-hooks")]
|
||||
use chrono::Utc;
|
||||
#[cfg(feature = "e2e-test-hooks")]
|
||||
use hmac::{Hmac, KeyInit, Mac};
|
||||
#[cfg(feature = "e2e-test-hooks")]
|
||||
use reqwest::StatusCode;
|
||||
#[cfg(feature = "e2e-test-hooks")]
|
||||
use rustfs::embedded::pause_embedded_startup_after_http_bind;
|
||||
use rustfs::embedded::{RustFSServerBuilder, find_available_port};
|
||||
#[cfg(feature = "e2e-test-hooks")]
|
||||
use sha2::{Digest, Sha256};
|
||||
#[cfg(feature = "e2e-test-hooks")]
|
||||
use std::time::Duration;
|
||||
|
||||
#[cfg(feature = "e2e-test-hooks")]
|
||||
type HmacSha256 = Hmac<Sha256>;
|
||||
|
||||
fn s3_client(endpoint: &str, access_key: &str, secret_key: &str) -> Client {
|
||||
let creds = Credentials::new(access_key, secret_key, None, None, "test");
|
||||
@@ -33,6 +48,62 @@ fn s3_client(endpoint: &str, access_key: &str, secret_key: &str) -> Client {
|
||||
Client::from_conf(config)
|
||||
}
|
||||
|
||||
#[cfg(feature = "e2e-test-hooks")]
|
||||
fn hex(bytes: impl AsRef<[u8]>) -> String {
|
||||
bytes.as_ref().iter().map(|byte| format!("{byte:02x}")).collect()
|
||||
}
|
||||
|
||||
#[cfg(feature = "e2e-test-hooks")]
|
||||
fn sha256_hex(bytes: &[u8]) -> String {
|
||||
hex(Sha256::digest(bytes))
|
||||
}
|
||||
|
||||
#[cfg(feature = "e2e-test-hooks")]
|
||||
fn hmac(key: &[u8], value: &str) -> Vec<u8> {
|
||||
let mut mac = HmacSha256::new_from_slice(key).expect("HMAC accepts arbitrary key lengths");
|
||||
mac.update(value.as_bytes());
|
||||
mac.finalize().into_bytes().to_vec()
|
||||
}
|
||||
|
||||
#[cfg(feature = "e2e-test-hooks")]
|
||||
fn signed_admin_request(
|
||||
client: &reqwest::Client,
|
||||
endpoint: &str,
|
||||
request_path: &str,
|
||||
access_key: &str,
|
||||
secret_key: &str,
|
||||
) -> reqwest::RequestBuilder {
|
||||
let host = endpoint
|
||||
.strip_prefix("http://")
|
||||
.or_else(|| endpoint.strip_prefix("https://"))
|
||||
.expect("embedded endpoint scheme");
|
||||
let payload_hash = sha256_hex(b"");
|
||||
let now = Utc::now();
|
||||
let amz_date = now.format("%Y%m%dT%H%M%SZ").to_string();
|
||||
let date = now.format("%Y%m%d").to_string();
|
||||
let canonical_headers = format!("host:{host}\nx-amz-content-sha256:{payload_hash}\nx-amz-date:{amz_date}\n");
|
||||
let signed_headers = "host;x-amz-content-sha256;x-amz-date";
|
||||
let (path, query) = request_path.split_once('?').unwrap_or((request_path, ""));
|
||||
let canonical_request = format!("GET\n{path}\n{query}\n{canonical_headers}\n{signed_headers}\n{payload_hash}");
|
||||
let scope = format!("{date}/us-east-1/s3/aws4_request");
|
||||
let string_to_sign = format!("AWS4-HMAC-SHA256\n{amz_date}\n{scope}\n{}", sha256_hex(canonical_request.as_bytes()));
|
||||
let date_key = hmac(format!("AWS4{secret_key}").as_bytes(), &date);
|
||||
let region_key = hmac(&date_key, "us-east-1");
|
||||
let service_key = hmac(®ion_key, "s3");
|
||||
let signing_key = hmac(&service_key, "aws4_request");
|
||||
let authorization = format!(
|
||||
"AWS4-HMAC-SHA256 Credential={access_key}/{scope}, SignedHeaders={signed_headers}, Signature={}",
|
||||
hex(hmac(&signing_key, &string_to_sign))
|
||||
);
|
||||
|
||||
client
|
||||
.get(format!("{endpoint}{request_path}"))
|
||||
.header("host", host)
|
||||
.header("x-amz-content-sha256", payload_hash)
|
||||
.header("x-amz-date", amz_date)
|
||||
.header("authorization", authorization)
|
||||
}
|
||||
|
||||
// backlog#1052 acceptance: a second embedded server in the same process no
|
||||
// longer aborts on write-once startup state — before this change,
|
||||
// `RustFSServer::build()` returned AlreadyStarted (guard) or panicked on
|
||||
@@ -246,3 +317,112 @@ async fn two_embedded_servers_isolate_auth_and_data_planes() {
|
||||
server_a.shutdown().await;
|
||||
server_b.shutdown().await;
|
||||
}
|
||||
|
||||
#[cfg(feature = "e2e-test-hooks")]
|
||||
#[tokio::test]
|
||||
async fn second_embedded_server_fails_closed_until_its_context_slot_is_installed() {
|
||||
let port_a = match find_available_port() {
|
||||
Ok(port) => port,
|
||||
Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return,
|
||||
Err(err) => panic!("find free port for server A: {err}"),
|
||||
};
|
||||
let server_a = RustFSServerBuilder::new()
|
||||
.address(format!("127.0.0.1:{port_a}"))
|
||||
.access_key("startup-window-access-a")
|
||||
.secret_key("startup-window-secret-a")
|
||||
.build()
|
||||
.await
|
||||
.expect("start embedded server A");
|
||||
let client_a = s3_client(&server_a.endpoint(), server_a.access_key(), server_a.secret_key());
|
||||
client_a
|
||||
.create_bucket()
|
||||
.bucket("startup-window")
|
||||
.send()
|
||||
.await
|
||||
.expect("server A creates the shared-name bucket");
|
||||
client_a
|
||||
.put_object()
|
||||
.bucket("startup-window")
|
||||
.key("marker.txt")
|
||||
.body(ByteStream::from_static(b"from A"))
|
||||
.send()
|
||||
.await
|
||||
.expect("server A writes its marker");
|
||||
|
||||
let port_b = match find_available_port() {
|
||||
Ok(port) => port,
|
||||
Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => {
|
||||
server_a.shutdown().await;
|
||||
return;
|
||||
}
|
||||
Err(err) => {
|
||||
server_a.shutdown().await;
|
||||
panic!("find free port for server B: {err}");
|
||||
}
|
||||
};
|
||||
let endpoint_b = format!("http://127.0.0.1:{port_b}");
|
||||
let b_access_key = "startup-window-access-b";
|
||||
let b_secret_key = "startup-window-secret-b";
|
||||
let mut barrier = pause_embedded_startup_after_http_bind(port_b);
|
||||
let startup_b = tokio::spawn(async move {
|
||||
RustFSServerBuilder::new()
|
||||
.address(format!("127.0.0.1:{port_b}"))
|
||||
.access_key(b_access_key)
|
||||
.secret_key(b_secret_key)
|
||||
.build()
|
||||
.await
|
||||
});
|
||||
tokio::time::timeout(Duration::from_secs(10), barrier.wait_until_http_bound())
|
||||
.await
|
||||
.expect("server B must bind HTTP before installing its context slot");
|
||||
|
||||
let http = reqwest::Client::builder()
|
||||
.no_proxy()
|
||||
.timeout(Duration::from_secs(5))
|
||||
.build()
|
||||
.expect("build local admin client without proxy");
|
||||
let inspect_path = "/rustfs/admin/v3/inspect-data?file=marker.txt&volume=startup-window";
|
||||
let before_install = signed_admin_request(&http, &endpoint_b, inspect_path, b_access_key, b_secret_key)
|
||||
.send()
|
||||
.await
|
||||
.expect("server B HTTP listener must accept the paused request");
|
||||
let before_install_status = before_install.status();
|
||||
let before_install_body = before_install.text().await.expect("read paused response body");
|
||||
assert_eq!(before_install_status, StatusCode::SERVICE_UNAVAILABLE, "{before_install_body}");
|
||||
assert!(
|
||||
before_install_body.contains("server context is not ready"),
|
||||
"paused request must not resolve server A: {before_install_body}"
|
||||
);
|
||||
|
||||
barrier.release();
|
||||
let server_b = tokio::time::timeout(Duration::from_secs(20), startup_b)
|
||||
.await
|
||||
.expect("server B startup must complete after releasing the barrier")
|
||||
.expect("server B startup task must not panic")
|
||||
.expect("start embedded server B");
|
||||
let client_b = s3_client(&server_b.endpoint(), server_b.access_key(), server_b.secret_key());
|
||||
client_b
|
||||
.create_bucket()
|
||||
.bucket("startup-window")
|
||||
.send()
|
||||
.await
|
||||
.expect("server B creates its isolated shared-name bucket");
|
||||
client_b
|
||||
.put_object()
|
||||
.bucket("startup-window")
|
||||
.key("marker.txt")
|
||||
.body(ByteStream::from_static(b"from B"))
|
||||
.send()
|
||||
.await
|
||||
.expect("server B writes its marker");
|
||||
|
||||
let after_install = signed_admin_request(&http, &server_b.endpoint(), inspect_path, b_access_key, b_secret_key)
|
||||
.send()
|
||||
.await
|
||||
.expect("server B admin request after context installation");
|
||||
assert_eq!(after_install.status(), StatusCode::OK);
|
||||
assert_eq!(after_install.bytes().await.expect("read server B marker"), b"from B".as_slice());
|
||||
|
||||
server_b.shutdown().await;
|
||||
server_a.shutdown().await;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user