From b4e0838b013771d98c138006b414a1fbb56b9062 Mon Sep 17 00:00:00 2001 From: overtrue Date: Sun, 6 Sep 2026 16:38:09 +0800 Subject: [PATCH] test(startup): observe pool repair classification and replicas --- crates/ecstore/src/core/pools.rs | 95 +++++++++++++++++++++++++++++++- crates/ecstore/src/store/init.rs | 23 ++++++-- 2 files changed, 112 insertions(+), 6 deletions(-) diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index dc941b20c..39463a3ab 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -5108,7 +5108,49 @@ async fn read_pool_meta_replicas(pools: Vec>, no_lock: bool) -> Vec "missing", + PoolMetaCasToken::Existing(_) => "existing", + PoolMetaCasToken::Unsafe => "unsafe", + }, + "etag": match &read.cas { PoolMetaCasToken::Existing(etag) => Some(etag), _ => None }, + }); + match &read.replica { + PoolMetaReplica::Valid { + raw, + canonical, + meta, + revision, + committed, + .. + } => { + observation["state"] = serde_json::json!("valid"); + observation["committed"] = serde_json::json!(committed); + observation["version"] = serde_json::json!(revision.version); + observation["cluster_id"] = serde_json::json!(revision.cluster_id); + observation["epoch"] = serde_json::json!(revision.epoch); + observation["generation"] = serde_json::json!(revision.generation); + observation["transaction_id"] = serde_json::json!(revision.transaction_id); + observation["pool_count"] = serde_json::json!(meta.pools.len()); + observation["payload_sha256"] = serde_json::json!(rustfs_utils::crypto::hex(Sha256::digest(canonical))); + observation["raw_sha256"] = serde_json::json!(rustfs_utils::crypto::hex(Sha256::digest(raw))); + } + PoolMetaReplica::Missing => observation["state"] = serde_json::json!("missing"), + PoolMetaReplica::Corrupt(_) => observation["state"] = serde_json::json!("corrupt"), + PoolMetaReplica::Incompatible(_) => observation["state"] = serde_json::json!("incompatible"), + PoolMetaReplica::Unreadable(_) => observation["state"] = serde_json::json!("unreadable"), + } + startup_cas_test_observe(observation); + } + } + reads } fn select_pool_meta_replicas_observing(write_state: &mut PoolMetaWriteState, replicas: Vec) -> Result @@ -5480,9 +5522,44 @@ fn pool_meta_cas_preconditions(token: &PoolMetaCasToken, object: &str) -> Result } } +#[cfg(feature = "e2e-test-hooks")] +struct StartupCasObservation { + attempt: uuid::Uuid, + phase: &'static str, + pools: Vec, +} + +#[cfg(feature = "e2e-test-hooks")] +tokio::task_local! { + static STARTUP_CAS_OBSERVATION: StartupCasObservation; +} + +// This scope follows only the directly polled startup future. Spawned work +// does not inherit it; receiver evidence retains its existing RPC tuple. +#[cfg(feature = "e2e-test-hooks")] +pub(crate) async fn startup_cas_test_scope( + attempt: uuid::Uuid, + phase: &'static str, + pools: &[Arc], + future: F, +) -> F::Output { + STARTUP_CAS_OBSERVATION + .scope( + StartupCasObservation { + attempt, + phase, + // These identities are never dereferenced or logged. The + // caller and operation keep the same pool Arcs alive. + pools: pools.iter().map(|pool| Arc::as_ptr(pool) as usize).collect(), + }, + future, + ) + .await +} + // Direct JSON diagnostics are independent of the startup tracing subscriber. #[cfg(feature = "e2e-test-hooks")] -fn startup_cas_test_observe(mut observation: serde_json::Value) { +pub(crate) fn startup_cas_test_observe(mut observation: serde_json::Value) { let Some(nonce) = std::env::var("RUSTFS_E2E_STARTUP_CAS_NONCE") .ok() .and_then(|value| uuid::Uuid::parse_str(&value).ok()) @@ -5491,6 +5568,10 @@ fn startup_cas_test_observe(mut observation: serde_json::Value) { }; observation["nonce"] = serde_json::json!(nonce); observation["pid"] = serde_json::json!(std::process::id()); + let _ = STARTUP_CAS_OBSERVATION.try_with(|scope| { + observation["attempt"] = serde_json::json!(scope.attempt); + observation["startup_phase"] = serde_json::json!(scope.phase); + }); let line = format!("RUSTFS_E2E_STARTUP_CAS {observation}\n"); let _ = std::io::Write::write_all(&mut std::io::stderr().lock(), line.as_bytes()); } @@ -5519,6 +5600,9 @@ where let observation = std::env::var_os("RUSTFS_E2E_STARTUP_CAS_NONCE").map(|_| { serde_json::json!({ "kind": "cas", "object": object, "phase": phase, + "pool": STARTUP_CAS_OBSERVATION.try_with(|scope| { + scope.pools.iter().position(|identity| *identity == Arc::as_ptr(&pool) as usize) + }).ok().flatten(), "payload_sha256": rustfs_utils::crypto::hex(Sha256::digest(&data)), "if_match": opts.http_preconditions.as_ref().and_then(|p| p.if_match.as_deref()), "if_none_match": opts.http_preconditions.as_ref().and_then(|p| p.if_none_match.as_deref()), @@ -5538,6 +5622,13 @@ where if let Some(mut observation) = observation { observation["ok"] = serde_json::json!(result.is_ok()); observation["etag"] = serde_json::json!(result.as_ref().ok().and_then(|info| info.etag.as_deref())); + observation["mod_time"] = serde_json::json!( + result + .as_ref() + .ok() + .and_then(|info| info.mod_time) + .map(|time| time.unix_timestamp_nanos().to_string()) + ); observation["error"] = serde_json::json!(result.as_ref().err().map(ToString::to_string)); startup_cas_test_observe(observation); } diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 86662f182..a38ba4077 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -630,14 +630,27 @@ impl ECStore { .pools .first() .is_some_and(|pool| pool_first_endpoint_is_local(&pool.endpoints)); + #[cfg(feature = "e2e-test-hooks")] + let startup_attempt = uuid::Uuid::new_v4(); let (meta, pool_meta_replica_state) = { let mut write_state = self.pool_meta_save_gate.lock().await; establish_pool_meta_bootstrap_identity_if_proven(self.pools.clone(), &mut write_state, should_persist_pool_meta) .await .map_err(|err| Error::other(format!("store init failed during establish_pool_meta_bootstrap_identity: {err}")))?; - load_pool_meta_for_startup(self.pools.clone(), &mut write_state).await? + let load = load_pool_meta_for_startup(self.pools.clone(), &mut write_state); + #[cfg(feature = "e2e-test-hooks")] + let load = crate::core::pools::startup_cas_test_scope(startup_attempt, "load", &self.pools, load); + load.await? }; let update = meta.validate(self.pools.clone())?; + #[cfg(feature = "e2e-test-hooks")] + crate::core::pools::startup_cas_test_observe(serde_json::json!({ + "kind": "startup-classifier", "attempt": startup_attempt, + "elected_writer": should_persist_pool_meta, + "needs_repair": pool_meta_replica_state.needs_repair, + "repair_write_safe": pool_meta_replica_state.repair_write_safe, + "topology_update": update, + })); let endpoints = runtime_sources::endpoint_pools_or_default(); let mut installed_pool_meta = if update { @@ -649,15 +662,17 @@ impl ECStore { // distributed startup can race on the same lock and replay the prior init bug. { let mut write_state = self.pool_meta_save_gate.lock().await; - installed_pool_meta = persist_pool_meta_for_startup_if_safe( + let persist = persist_pool_meta_for_startup_if_safe( &installed_pool_meta, self.pools.clone(), pool_meta_replica_state, &mut write_state, update, should_persist_pool_meta, - ) - .await?; + ); + #[cfg(feature = "e2e-test-hooks")] + let persist = crate::core::pools::startup_cas_test_scope(startup_attempt, "persist", &self.pools, persist); + installed_pool_meta = persist.await?; } {