mirror of
https://github.com/rustfs/rustfs.git
synced 2026-09-06 20:19:14 +00:00
test(rpc): retain bootstrap authority across delayed requests
This commit is contained in:
overtrue
@@ -491,6 +491,90 @@ enum LocalMutationTarget {
|
||||
Unbound,
|
||||
}
|
||||
|
||||
#[cfg(feature = "e2e-test-hooks")]
|
||||
pub(crate) mod rename_target_capture_test_hook {
|
||||
use super::LocalMutationTarget;
|
||||
use rustfs_protos::proto_gen::node_service::RenameDataRequest;
|
||||
use std::sync::{LazyLock, Mutex};
|
||||
use tokio::sync::oneshot;
|
||||
use uuid::Uuid;
|
||||
|
||||
struct Hook {
|
||||
id: Uuid,
|
||||
disk: String,
|
||||
volume: String,
|
||||
path: String,
|
||||
captured: oneshot::Sender<bool>,
|
||||
release: oneshot::Receiver<()>,
|
||||
}
|
||||
|
||||
static HOOK: LazyLock<Mutex<Option<Hook>>> = LazyLock::new(|| Mutex::new(None));
|
||||
|
||||
/// One exact signed rename paused after its listener target was captured.
|
||||
/// Dropping the handle removes an unused hook and releases an entered one.
|
||||
pub struct RenameTargetCapturePause {
|
||||
id: Uuid,
|
||||
captured: oneshot::Receiver<bool>,
|
||||
release: Option<oneshot::Sender<()>>,
|
||||
}
|
||||
|
||||
impl RenameTargetCapturePause {
|
||||
pub async fn wait_until_captured(&mut self) -> bool {
|
||||
(&mut self.captured)
|
||||
.await
|
||||
.expect("matching rename must report its actual captured target")
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for RenameTargetCapturePause {
|
||||
fn drop(&mut self) {
|
||||
let unused = HOOK
|
||||
.lock()
|
||||
.expect("rename capture hook lock")
|
||||
.take_if(|hook| hook.id == self.id);
|
||||
drop(unused);
|
||||
if let Some(release) = self.release.take() {
|
||||
let _ = release.send(());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub fn pause_rename_after_target_capture(disk: &str, volume: &str, path: &str) -> RenameTargetCapturePause {
|
||||
let id = Uuid::new_v4();
|
||||
let (captured_tx, captured) = oneshot::channel();
|
||||
let (release, release_rx) = oneshot::channel();
|
||||
let mut active = HOOK.lock().expect("rename capture hook lock");
|
||||
if active.is_some() {
|
||||
drop(active);
|
||||
panic!("only one rename capture hook may be active");
|
||||
}
|
||||
*active = Some(Hook {
|
||||
id,
|
||||
disk: disk.to_owned(),
|
||||
volume: volume.to_owned(),
|
||||
path: path.to_owned(),
|
||||
captured: captured_tx,
|
||||
release: release_rx,
|
||||
});
|
||||
RenameTargetCapturePause {
|
||||
id,
|
||||
captured,
|
||||
release: Some(release),
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) async fn wait(target: &LocalMutationTarget, request: &RenameDataRequest) {
|
||||
let hook = {
|
||||
let mut active = HOOK.lock().expect("rename capture hook lock");
|
||||
active.take_if(|hook| hook.disk == request.disk && hook.volume == request.dst_volume && hook.path == request.dst_path)
|
||||
};
|
||||
if let Some(hook) = hook {
|
||||
let _ = hook.captured.send(matches!(target, LocalMutationTarget::Bootstrap(_)));
|
||||
let _ = hook.release.await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl std::fmt::Debug for NodeService {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
f.debug_struct("NodeService")
|
||||
|
||||
@@ -1251,6 +1251,8 @@ impl NodeService {
|
||||
)?;
|
||||
let request = request.into_inner();
|
||||
let target = self.local_mutation_target();
|
||||
#[cfg(feature = "e2e-test-hooks")]
|
||||
super::rename_target_capture_test_hook::wait(&target, &request).await;
|
||||
let decoded_file_info = match decode_rename_data_request_file_info(&request.file_info_bin, &request.file_info) {
|
||||
Ok(file_info) => file_info,
|
||||
Err(err) => {
|
||||
|
||||
@@ -18,3 +18,9 @@ pub(crate) use crate::storage::rpc::node_service::{heal::heal_topology_fingerpri
|
||||
pub(crate) use crate::storage::rpc::node_service::{make_scanner_control_server, make_server_for_slot};
|
||||
pub use crate::storage::rpc::{make_heal_control_server, make_server, make_tier_mutation_control_server};
|
||||
pub type NodeService = crate::storage::rpc::NodeService;
|
||||
|
||||
#[cfg(feature = "e2e-test-hooks")]
|
||||
#[doc(hidden)]
|
||||
pub use crate::storage::rpc::node_service::rename_target_capture_test_hook::{
|
||||
RenameTargetCapturePause, pause_rename_after_target_capture,
|
||||
};
|
||||
|
||||
@@ -890,4 +890,180 @@ mod signed_target_rpc {
|
||||
timeout(WAIT, server_a.shutdown()).await.expect("bounded A shutdown");
|
||||
timeout(WAIT, server_b.shutdown()).await.expect("bounded B shutdown");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn signed_bootstrap_request_does_not_upgrade_after_context_installation() {
|
||||
common::run_embedded_test(|| async {
|
||||
timeout(WAIT * 6, signed_delayed_bootstrap_body())
|
||||
.await
|
||||
.expect("bounded delayed Bootstrap fixture");
|
||||
});
|
||||
}
|
||||
|
||||
async fn signed_delayed_bootstrap_body() {
|
||||
use rustfs::storage::tonic_service::pause_rename_after_target_capture;
|
||||
|
||||
let root_b = tempfile::tempdir().expect("B root");
|
||||
let server_b = timeout(
|
||||
WAIT,
|
||||
RustFSServerBuilder::new()
|
||||
.address(format!("127.0.0.1:{}", find_available_port().expect("B port")))
|
||||
.volume(root_b.path().to_str().expect("B path"))
|
||||
.access_key("delayed-bootstrap-access")
|
||||
.secret_key("delayed-bootstrap-secret")
|
||||
.build(),
|
||||
)
|
||||
.await
|
||||
.expect("bounded B startup")
|
||||
.expect("start global B");
|
||||
let global_b = resolve_object_store_handle().expect("B's published context");
|
||||
let disk_b = local_fixture_disk(root_b.path()).await;
|
||||
let endpoints = global_b.instance_endpoints().expect("B instance topology");
|
||||
let paths: Vec<_> = endpoints
|
||||
.0
|
||||
.iter()
|
||||
.flat_map(|pool| pool.endpoints.as_ref().iter())
|
||||
.map(ToString::to_string)
|
||||
.collect();
|
||||
assert_eq!(paths, [disk_b.endpoint().to_string()], "the ambient store owns B");
|
||||
stage(&disk_b, USER_VOLUME, "delayed-sentinel", b"B-is-not-the-listener-target").await;
|
||||
let sentinel_path = root_b.path().join(USER_VOLUME).join("delayed-sentinel/xl.meta");
|
||||
let sentinel_before = tokio::fs::read(&sentinel_path).await.expect("B sentinel bytes");
|
||||
|
||||
let root_a = tempfile::tempdir().expect("A root");
|
||||
let port_a = find_available_port().expect("A port");
|
||||
let address_a = format!("127.0.0.1:{port_a}").parse().expect("A address");
|
||||
let mut startup_barrier = Some(pause_embedded_startup_after_http_bind(port_a));
|
||||
let startup_a = RustFSServerBuilder::new()
|
||||
.address(format!("127.0.0.1:{port_a}"))
|
||||
.volume(root_a.path().to_str().expect("A path"))
|
||||
.access_key("delayed-bootstrap-access")
|
||||
.secret_key("delayed-bootstrap-secret")
|
||||
.build();
|
||||
tokio::pin!(startup_a);
|
||||
timeout(WAIT, async {
|
||||
tokio::select! {
|
||||
() = startup_barrier.as_mut().expect("startup barrier").wait_until_http_bound() => {}
|
||||
startup = startup_a.as_mut() => {
|
||||
let _unexpected_server = startup.expect("A initial startup");
|
||||
panic!("A must reach its pre-AppContext barrier");
|
||||
}
|
||||
}
|
||||
})
|
||||
.await
|
||||
.expect("bounded A listener startup");
|
||||
|
||||
let disk_a = local_fixture_disk(root_a.path()).await;
|
||||
let delayed_info = stage(&disk_a, USER_VOLUME, "delayed-source", b"captured-Bootstrap-must-not-publish").await;
|
||||
let control_info = stage(&disk_a, USER_VOLUME, "control-source", b"new-Ready-request-can-publish").await;
|
||||
let source_path = root_a.path().join(USER_VOLUME).join("delayed-source/xl.meta");
|
||||
let destination_path = root_a.path().join(USER_VOLUME).join("delayed-destination/xl.meta");
|
||||
let source_before = tokio::fs::read(&source_path).await.expect("delayed source bytes");
|
||||
let mut connection = SingleConnection::connect(address_a).await;
|
||||
let mut server_a = None;
|
||||
let mut startup_finished = false;
|
||||
|
||||
let (observations, delayed_result, source_after, destination_exists, sentinel_after) = {
|
||||
let mut capture =
|
||||
pause_rename_after_target_capture(&disk_a.endpoint().to_string(), USER_VOLUME, "delayed-destination");
|
||||
let mut delayed_client = connection.client.clone();
|
||||
let delayed = delayed_client.rename_data(signed_rename(
|
||||
&disk_a,
|
||||
USER_VOLUME,
|
||||
"delayed-source",
|
||||
"delayed-destination",
|
||||
&delayed_info,
|
||||
));
|
||||
tokio::pin!(delayed);
|
||||
let mut early_response = None;
|
||||
|
||||
// Bound all work while the request is parked to less than the
|
||||
// existing channel's 30-second deadline; no timeout is disabled.
|
||||
let observations = std::panic::AssertUnwindSafe(timeout(Duration::from_secs(20), async {
|
||||
let was_bootstrap = tokio::select! {
|
||||
observed = capture.wait_until_captured() => observed,
|
||||
response = delayed.as_mut() => {
|
||||
early_response = Some(response);
|
||||
panic!("signed request finished before the capture pause: {early_response:?}");
|
||||
},
|
||||
};
|
||||
assert!(was_bootstrap, "the actual authenticated handler captured Bootstrap");
|
||||
assert!(Arc::ptr_eq(&global_b, &resolve_object_store_handle().expect("global B")));
|
||||
assert_eq!(tokio::fs::read(&source_path).await.expect("source before install"), source_before);
|
||||
assert!(!destination_path.exists());
|
||||
|
||||
startup_barrier.take().expect("unreleased startup barrier").release();
|
||||
let started = startup_a.as_mut().await;
|
||||
startup_finished = true;
|
||||
server_a = Some(started.expect("normal A context installation"));
|
||||
assert!(Arc::ptr_eq(&global_b, &resolve_object_store_handle().expect("global remains B")));
|
||||
assert_eq!(connection.peer, server_a.as_ref().expect("A handle").address());
|
||||
|
||||
// A separate source prevents this control from consuming the
|
||||
// delayed request's data and masking an erroneous second lookup.
|
||||
let ready = connection
|
||||
.rename(signed_rename(&disk_a, USER_VOLUME, "control-source", "ready-control", &control_info))
|
||||
.await;
|
||||
assert!(ready.success, "a fresh signed user request must actually use Ready: {:?}", ready.error);
|
||||
assert_body(&disk_a, USER_VOLUME, "ready-control", &control_info).await;
|
||||
assert_body(&disk_a, USER_VOLUME, "delayed-source", &delayed_info).await;
|
||||
assert!(!destination_path.exists(), "the original request remains parked");
|
||||
connection.assert_original_connection();
|
||||
}))
|
||||
.catch_unwind()
|
||||
.await;
|
||||
|
||||
// Release on every assertion/timeout path, then drain the original
|
||||
// RPC before shutting down the server and its connection.
|
||||
drop(capture);
|
||||
if let Some(barrier) = startup_barrier.take() {
|
||||
barrier.release();
|
||||
}
|
||||
if !startup_finished {
|
||||
let started = timeout(WAIT, startup_a.as_mut()).await;
|
||||
if let Ok(Ok(started)) = started {
|
||||
server_a = Some(started);
|
||||
}
|
||||
}
|
||||
let delayed_result = match early_response {
|
||||
Some(response) => Ok(response),
|
||||
None => timeout(WAIT, delayed.as_mut()).await,
|
||||
};
|
||||
let source_after = tokio::fs::read(&source_path).await;
|
||||
let destination_exists = tokio::fs::try_exists(&destination_path).await;
|
||||
let sentinel_after = tokio::fs::read(&sentinel_path).await;
|
||||
(observations, delayed_result, source_after, destination_exists, sentinel_after)
|
||||
};
|
||||
let connection_attempts = connection.attempts.load(Ordering::SeqCst);
|
||||
drop(connection);
|
||||
let shutdown_a = if let Some(server) = server_a {
|
||||
Some(timeout(WAIT, server.shutdown()).await)
|
||||
} else {
|
||||
None
|
||||
};
|
||||
let shutdown_b = timeout(WAIT, server_b.shutdown()).await;
|
||||
if let Some(result) = shutdown_a {
|
||||
result.expect("bounded A shutdown");
|
||||
}
|
||||
shutdown_b.expect("bounded B shutdown");
|
||||
|
||||
assert_eq!(connection_attempts, 1, "the original channel must not redial");
|
||||
match observations {
|
||||
Err(panic) => std::panic::resume_unwind(panic),
|
||||
Ok(result) => result.expect("complete capture/install/Ready-control within the parked request deadline"),
|
||||
}
|
||||
let response = delayed_result
|
||||
.expect("bounded original request drain")
|
||||
.expect("the original signed request must return an application result")
|
||||
.into_inner();
|
||||
assert!(
|
||||
!response.success,
|
||||
"a captured Bootstrap request must not upgrade to Ready after its await"
|
||||
);
|
||||
let error: DiskError = response.error.expect("typed Bootstrap rejection").into();
|
||||
assert_eq!(error, DiskError::FileAccessDenied);
|
||||
assert_eq!(source_after.expect("original source remains readable"), source_before);
|
||||
assert!(!destination_exists.expect("read original destination state"));
|
||||
assert_eq!(sentinel_after.expect("global B sentinel survives"), sentinel_before);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user