From bb6d55a99f8fffdf4e20b3a809162d387ffc42b4 Mon Sep 17 00:00:00 2001 From: discord9 Date: Fri, 7 Aug 2026 16:40:46 +0800 Subject: [PATCH] feat: support old-stage datanode config overlays (#8647) * feat: support old-stage datanode config overlays Signed-off-by: discord9 * fix: derive compat overlay policy from WAL config Signed-off-by: discord9 --------- Signed-off-by: discord9 --- tests/compatibility/AGENTS.md | 37 +- tests/compatibility/README.md | 42 +- tests/runner/src/cmd.rs | 1 + tests/runner/src/cmd/compat.rs | 787 +++++++++++++++++++---- tests/runner/src/cmd/compat_case.rs | 93 +++ tests/runner/src/cmd/datanode_overlay.rs | 637 ++++++++++++++++++ tests/runner/src/env/bare.rs | 214 +++++- tests/runner/src/server_mode.rs | 176 ++++- tests/runner/src/util.rs | 62 +- 9 files changed, 1870 insertions(+), 179 deletions(-) create mode 100644 tests/runner/src/cmd/datanode_overlay.rs diff --git a/tests/compatibility/AGENTS.md b/tests/compatibility/AGENTS.md index 8ef9b6c011..144f5af332 100644 --- a/tests/compatibility/AGENTS.md +++ b/tests/compatibility/AGENTS.md @@ -28,12 +28,42 @@ This file is intended for AI agents editing compat cases or the compat runner. F - All required fields must be non-empty: `name`, `reason`, `introduced_by`, `topologies`, `from_range`, `to_range`, `features`, `owner`. +## Old-Stage Datanode Overlays + +- An old-stage datanode overlay is declared only as: + + ```toml + [old_config] + datanode = "old-datanode.overlay.toml" + ``` + + `datanode` is required if `[old_config]` exists. Empty tables and unknown + keys are parse errors. +- The sidecar path is relative to its case directory and must remain confined + there. It contains native datanode TOML and is loaded and preflighted before + services or state are created. +- The runner merges tables recursively only when both values are tables. + Scalars, type mismatches, arrays, and arrays of tables replace atomically; + `region_engine` has no special merge behavior. +- Do not use an overlay to set runner-owned fields: `mode`, `node_id`, + `storage.data_home`, `meta_client_options.metasrv_addrs`, `wal.provider`, or + `wal.dir` for Raft WAL / `wal.broker_endpoints` for Kafka WAL. The runner + restores or deletes them according to its baseline and warns without showing + values. + ## Phase Semantics - `setup.sql` runs on the **old (from)** binary. Only success is required; output is not compared to any file. - `verify.sql` runs on the **new (to)** binary. Output is compared against `verify.result`. +- Overlays apply only to old-stage datanodes and survive old-stage setup + restarts; the current stage uses a clean configuration. +- The baseline profile runs first. Cases with semantically equivalent datanode + TOML share one sequential, isolated profile, and every profile has an + independent state and etcd lifecycle. +- Without fail-fast, only cases with successful setup are verified. Fail-fast + cleans up the active profile before stopping. ## PostgreSQL Protocol Cases @@ -48,9 +78,10 @@ cargo run -p sqlness-runner -- compat --dry-run [--from-version vX.Y.Z] [--test- ``` The dry-run performs full discovery and filtering (name, topology, metadata -validation, namespace dedup, version-range matching) but starts no services, -creates no temp dirs, and mutates no files. Use it to check which cases -would be selected before a real run. +validation, namespace dedup, version-range matching) and displays selected +profiles, cases, and sidecar paths without configuration values. It starts no +services, creates no temp dirs, and mutates no files. Use it to check which +cases would be selected before a real run. ## CI Version Window diff --git a/tests/compatibility/README.md b/tests/compatibility/README.md index a88fddeaf7..542f10debf 100644 --- a/tests/compatibility/README.md +++ b/tests/compatibility/README.md @@ -68,6 +68,34 @@ namespace = "my_explicit_namespace" # defaults to sanitized directory name **Required fields**: `name`, `reason`, `introduced_by`, `topologies`, `from_range`, `to_range`, `features`, `owner`. +### Old-Stage Datanode Configuration Overlay + +To apply a datanode configuration overlay while running the old stage, add this +strict optional table to `case.toml`: + +```toml +[old_config] +datanode = "old-datanode.overlay.toml" +``` + +`datanode` is required whenever `[old_config]` is present; empty tables and +unknown keys are rejected. The reference is relative to the case directory and +must remain confined to that directory. The sidecar is native datanode TOML, +which the runner loads and preflights before starting services or creating +state. + +The runner first applies the datanode baseline, then merges the sidecar. Tables +merge recursively only when both values are tables. Scalars, type mismatches, +arrays, and arrays of tables replace the baseline value atomically. In +particular, `region_engine` has no special merge behavior. + +Runner-owned fields cannot be changed by an overlay: `mode`, `node_id`, +`storage.data_home`, `meta_client_options.metasrv_addrs`, and `wal.provider`, +plus `wal.dir` for Raft WAL or `wal.broker_endpoints` for Kafka WAL. The runner +restores these fields to its baseline values, or deletes them when the baseline +has no value. It warns about protected-field overrides without printing their +values. + ### Version-Range Filtering `from_range` and `to_range` control which binary versions a case applies to: @@ -166,9 +194,19 @@ Each case runs in its own database namespace to prevent cross-case interference: ## Batch Behavior -- All cases in a run share one cluster lifecycle: start from-version cluster → run all setups → restart with to-version binary → run all verifies -- Cases run **serially** (no parallelism in PR1). Namespace state is session/protocol state and cannot be shared concurrently. +- The baseline (no-overlay) profile runs first. Cases whose old datanode TOML + is semantically equivalent share one profile; profiles run serially and in + isolation. +- Each profile has its own state and etcd lifecycle. Its overlay is applied + only to old-stage datanodes and remains in effect through old-stage setup + restarts. The current stage always uses a clean configuration. +- Cases run **serially** (no parallelism in PR1). Namespace state is + session/protocol state and cannot be shared concurrently. - Same namespace across cases is rejected. +- Without fail-fast, the runner verifies only cases whose setup succeeded. + With fail-fast, it cleans up the active profile before stopping. +- `--dry-run` displays the selected profiles, cases, and sidecar paths without + printing configuration values; it starts no services. ## xfail Policy (Future) diff --git a/tests/runner/src/cmd.rs b/tests/runner/src/cmd.rs index e17d910b85..c0fed551a3 100644 --- a/tests/runner/src/cmd.rs +++ b/tests/runner/src/cmd.rs @@ -15,6 +15,7 @@ pub(crate) mod bare; pub(crate) mod compat; pub(crate) mod compat_case; +pub(crate) mod datanode_overlay; pub(crate) mod kube; use std::path::PathBuf; diff --git a/tests/runner/src/cmd/compat.rs b/tests/runner/src/cmd/compat.rs index ce33673a4a..b7473131ef 100644 --- a/tests/runner/src/cmd/compat.rs +++ b/tests/runner/src/cmd/compat.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::collections::BTreeMap; use std::path::PathBuf; use std::sync::Arc; @@ -22,7 +23,10 @@ use sqlness::interceptor::{InterceptorRef, Registry}; use crate::cmd::bare::ServerAddr; use crate::cmd::compat_case::{self, CompatCase, try_infer_version, version_matches_range}; -use crate::env::bare::{Env, StoreConfig, WalConfig}; +use crate::cmd::datanode_overlay::{ + DatanodeOverlay, DatanodeProtectionPolicy, PreparedDatanodeOverlay, +}; +use crate::env::bare::{Env, GreptimeDB, StoreConfig, WalConfig}; use crate::protocol_interceptor::{self, POSTGRES, PROTOCOL_KEY}; use crate::util; @@ -326,16 +330,12 @@ impl CompatCommand { ); } - if let Some(expected_cases) = self.expect_cases { - assert_eq!( - cases.len(), - expected_cases, - "Expected {expected_cases} compatibility cases after all filtering, found {}", - cases.len() - ); - } - if cases.is_empty() { + if let Some(expected_cases) = self.expect_cases { + panic!( + "Expected {expected_cases} compatibility cases after all filtering, found 0" + ); + } if dry_run { println!("DRY-RUN: no compat cases would run after version-range filtering"); } else { @@ -344,8 +344,25 @@ impl CompatCommand { return; } + let wal_config = WalConfig::RaftEngine; + let profiles = + prepare_compat_profiles(cases, &wal_config).unwrap_or_else(|error| panic!("{error}")); + emit_protected_path_warnings(&profiles); + + if let Some(expected_cases) = self.expect_cases { + assert_eq!( + profiles.case_count(), + expected_cases, + "Expected {expected_cases} compatibility cases after all filtering, found {}", + profiles.case_count() + ); + } + if dry_run { - println!("DRY-RUN: would run {} compat case(s)", cases.len()); + println!( + "DRY-RUN: would run {} compat case(s)", + profiles.case_count() + ); println!(" topology: {}", topology.as_str()); println!( " from version: {}", @@ -359,22 +376,30 @@ impl CompatCommand { .as_deref() .unwrap_or("unknown (use --to-bins-dir or build debug binary)") ); - if pre_filter_count != cases.len() { + if pre_filter_count != profiles.case_count() { println!(); println!( "Version-range filtering reduced {} → {} cases (see 'Skipping case' messages above)", pre_filter_count, - cases.len() + profiles.case_count() ); } println!(); - for c in &cases { + for c in profiles.cases() { println!(" case: {}", c.metadata.name); println!(" namespace: {}", c.namespace); println!(" from_range: {:?}", c.metadata.from_range); println!(" to_range: {:?}", c.metadata.to_range); println!(" features: {:?}", c.metadata.features); } + println!("DRY-RUN: compatibility profiles:"); + for profile in profiles.iter() { + println!(" profile: {}", profile.profile_id()); + println!(" cases: {}", profile.case_names().join(", ")); + for source in profile.sources() { + println!(" sidecar: {}", source.display()); + } + } println!(); println!("Dry run complete. Remove --dry-run to execute."); return; @@ -382,133 +407,499 @@ impl CompatCommand { println!( "Running {} compat case(s) with topology {}:", - cases.len(), + profiles.case_count(), topology.as_str() ); - for c in &cases { + for c in profiles.cases() { println!( " - {} (namespace: {}, topologies: {:?})", c.metadata.name, c.namespace, c.metadata.topologies ); } - // ---- 6. Create temp directory (after filtering so early exits don't leave empty dirs) ---- - let temp_dir = tempfile::Builder::new() - .prefix("sqlness-compat") - .tempdir() - .unwrap(); - let sqlness_home = temp_dir.keep(); - unsafe { - std::env::set_var( - "SQLNESS_HOME", - sqlness_home.join("copy").display().to_string(), - ); - } - - // ---- 7. Build interceptor registry ---- + // ---- 6. Build interceptor registry ---- let interceptor_registry = create_interceptor_registry(); - - // ---- 7b. Create Env for the selected topology ---- - let setup_etcd = topology == CompatTopology::Distributed && self.setup_etcd; - let store_config = StoreConfig { - store_addrs: if setup_etcd { - vec!["127.0.0.1:2379".to_string()] - } else { - vec![] - }, - setup_etcd, - setup_pg: None, - setup_mysql: None, - enable_flat_format: false, - enable_gc: false, - }; - - let env = Env::new( - sqlness_home.clone(), - ServerAddr::default(), - WalConfig::RaftEngine, - self.pull_version_on_need, + let to_bins_dir = to_bins_dir.expect("to_bins_dir must be resolved in non-dry-run mode"); + let profile_config = ProfileRunConfig { + interceptor_registry: &interceptor_registry, from_bins_dir, - store_config, - vec![], - ); - - // ---- 7c. Etcd cleanup guard ---- - // Arm this only immediately before starting the cluster. Earlier validation - // failures should not stop an unrelated local container named `etcd`. - let mut etcd_guard = if setup_etcd { - Some(EtcdGuard::new()) - } else { - None + to_bins_dir, + pull_version_on_need: self.pull_version_on_need, + setup_etcd: self.setup_etcd, + fail_fast: self.fail_fast, + topology, + preserve_state: self.preserve_state, + wal_config, }; - - // ---- 8. Run setup phase on the from-version cluster ---- - println!("Starting from-version {} cluster...", topology.as_str()); - let mut db = match topology { - CompatTopology::Distributed => env.compat_start_distributed(0).await, - CompatTopology::Standalone => env.compat_start_standalone(0).await, - }; - - println!("Running setup phase..."); - for case in &cases { - run_compat_phase(&db, case, &interceptor_registry, CompatPhase::Setup) - .await - .unwrap_or_else(|e| panic!("Setup failed for case '{}': {e}", case.metadata.name)); - println!(" Setup: {} - OK", case.metadata.name); + let mut failures = Vec::new(); + let mut preserved_states = Vec::new(); + for profile in profiles.iter() { + let outcome = run_profile(profile, &profile_config).await; + failures.extend(outcome.failures); + if let Some(state) = outcome.preserved_state { + preserved_states.push(state); + } + if outcome.stop_remaining_profiles { + break; + } } - // ---- 9. Switch to the to-version binary and restart cluster ---- - // to_bins_dir was already resolved during version-range filtering - println!("Restarting cluster with to-version binary on preserved state..."); - let to_bins_dir = to_bins_dir.expect("to_bins_dir must be resolved in non-dry-run mode"); - env.compat_restart(&db, to_bins_dir).await; + for state in &preserved_states { + println!("Preserved compat profile state: {}", state.display()); + } + + if failures.is_empty() { + println!("\n\x1b[32mAll compat tests passed!\x1b[0m"); + } else { + panic!("\n\x1b[31mFailed cases: {}\x1b[0m", failures.join(", ")); + } + } +} + +/// Cases sharing one old-stage datanode configuration lifecycle. +#[derive(Debug)] +enum CompatProfile { + Baseline { + cases: Vec, + }, + Overlay { + overlay: Arc, + cases: Vec, + sources: Vec, + }, +} + +impl CompatProfile { + fn profile_id(&self) -> &str { + match self { + Self::Baseline { .. } => "baseline", + Self::Overlay { overlay, .. } => overlay.profile_id(), + } + } + + fn cases(&self) -> &[CompatCase] { + match self { + Self::Baseline { cases } | Self::Overlay { cases, .. } => cases, + } + } + + fn case_names(&self) -> Vec<&str> { + self.cases() + .iter() + .map(|case| case.metadata.name.as_str()) + .collect() + } + + fn sources(&self) -> &[PathBuf] { + match self { + Self::Baseline { .. } => &[], + Self::Overlay { sources, .. } => sources, + } + } + + fn overlay(&self) -> Option> { + match self { + Self::Baseline { .. } => None, + Self::Overlay { overlay, .. } => Some(Arc::clone(overlay)), + } + } + + fn touched_protected_paths(&self) -> &[crate::cmd::datanode_overlay::DottedPath] { + match self { + Self::Baseline { .. } => &[], + Self::Overlay { overlay, .. } => overlay.touched_protected_paths(), + } + } +} + +/// Deterministically ordered compatibility profiles. +#[derive(Debug)] +struct CompatProfiles(Vec); + +impl CompatProfiles { + fn iter(&self) -> impl Iterator { + self.0.iter() + } + + fn cases(&self) -> impl Iterator { + self.0.iter().flat_map(CompatProfile::cases) + } + + fn case_count(&self) -> usize { + self.0.iter().map(|profile| profile.cases().len()).sum() + } +} + +#[derive(Debug)] +struct OverlayProfileGroup { + overlay: Arc, + cases: Vec, + sources: Vec, +} + +fn prepare_compat_profiles( + cases: Vec, + wal_config: &WalConfig, +) -> Result { + let protection = DatanodeProtectionPolicy::for_wal(wal_config); + let mut baseline_cases = Vec::new(); + let mut overlays: BTreeMap<[u8; 32], OverlayProfileGroup> = BTreeMap::new(); + + for case in cases { + let Some(reference) = case.metadata.old_datanode_overlay() else { + baseline_cases.push(case); + continue; + }; + let overlay = DatanodeOverlay::load(&case.dir, reference).map_err(|error| { + format!( + "Failed to load old datanode overlay for compatibility case '{}': {error}", + case.metadata.name + ) + })?; + let prepared = Arc::new(overlay.prepare(&protection).map_err(|error| { + format!( + "Failed to prepare old datanode overlay for compatibility case '{}': {error}", + case.metadata.name + ) + })?); + let key = *prepared.profile_key(); + let source = prepared.source().to_path_buf(); + let entry = overlays.entry(key).or_insert_with(|| OverlayProfileGroup { + overlay: Arc::clone(&prepared), + cases: Vec::new(), + sources: Vec::new(), + }); + entry.cases.push(case); + entry.sources.push(source); + } + + baseline_cases.sort_by(|left, right| left.metadata.name.cmp(&right.metadata.name)); + let mut profiles = Vec::new(); + if !baseline_cases.is_empty() { + profiles.push(CompatProfile::Baseline { + cases: baseline_cases, + }); + } + for ( + _, + OverlayProfileGroup { + overlay, + mut cases, + mut sources, + }, + ) in overlays + { + cases.sort_by(|left, right| left.metadata.name.cmp(&right.metadata.name)); + sources.sort(); + sources.dedup(); + profiles.push(CompatProfile::Overlay { + overlay, + cases, + sources, + }); + } + + Ok(CompatProfiles(profiles)) +} + +fn emit_protected_path_warnings(profiles: &CompatProfiles) { + for profile in profiles.iter() { + let paths: Vec<_> = profile + .touched_protected_paths() + .iter() + .map(ToString::to_string) + .collect(); + if paths.is_empty() { + continue; + } + println!( + "{}", + format_protected_path_warning(profile.profile_id(), profile.case_names(), paths) + ); + } +} + +fn format_protected_path_warning( + profile_id: &str, + mut case_names: Vec<&str>, + mut paths: Vec, +) -> String { + case_names.sort_unstable(); + paths.sort_unstable(); + format!( + "Warning: datanode overlay profile {profile_id} touches runner-owned paths [{}] for cases [{}]", + paths.join(", "), + case_names.join(", ") + ) +} + +#[derive(Debug)] +struct ProfileOutcome { + failures: Vec, + stop_remaining_profiles: bool, + preserved_state: Option, +} + +/// Pure compatibility profile progress policy, independent of process ownership. +#[derive(Default)] +struct ProfileProgress { + successful_setup_indexes: Vec, + failures: Vec, + fail_fast_setup_failure: bool, + fail_fast_verify_failure: bool, + cleanup_failed: bool, +} + +impl ProfileProgress { + fn record_setup_success(&mut self, case_index: usize) { + self.successful_setup_indexes.push(case_index); + } + + fn record_setup_failure(&mut self, failure: String, fail_fast: bool) -> bool { + self.failures.push(failure); + self.fail_fast_setup_failure = fail_fast; + fail_fast + } + + fn record_verify_failure(&mut self, failure: String, fail_fast: bool) -> bool { + self.failures.push(failure); + self.fail_fast_verify_failure = fail_fast; + fail_fast + } + + fn record_cleanup_failure(&mut self, failure: String) { + self.failures.push(failure); + self.cleanup_failed = true; + } + + fn should_transition_to_current(&self) -> bool { + !self.successful_setup_indexes.is_empty() && !self.fail_fast_setup_failure + } + + fn should_stop_remaining_profiles(&self) -> bool { + self.fail_fast_setup_failure || self.fail_fast_verify_failure || self.cleanup_failed + } +} + +struct ProfileRunConfig<'a> { + interceptor_registry: &'a Registry, + from_bins_dir: Option, + to_bins_dir: PathBuf, + pull_version_on_need: bool, + setup_etcd: bool, + fail_fast: bool, + topology: CompatTopology, + preserve_state: bool, + wal_config: WalConfig, +} + +async fn run_profile(profile: &CompatProfile, config: &ProfileRunConfig<'_>) -> ProfileOutcome { + println!("Running compatibility profile {}", profile.profile_id()); + let temp_dir = tempfile::Builder::new() + .prefix(&format!("sqlness-compat-{}-", profile.profile_id())) + .tempdir() + .unwrap(); + let profile_state = ProfileStateGuard::new(temp_dir, config.preserve_state); + let sqlness_home = profile_state.path().to_path_buf(); + // Standalone runs need no etcd. + let setup_etcd = config.topology == CompatTopology::Distributed && config.setup_etcd; + unsafe { + std::env::set_var( + "SQLNESS_HOME", + sqlness_home.join("copy").display().to_string(), + ); + } + + let store_config = StoreConfig { + store_addrs: setup_etcd + .then(|| "127.0.0.1:2379".to_string()) + .into_iter() + .collect(), + setup_etcd, + setup_pg: None, + setup_mysql: None, + enable_flat_format: false, + enable_gc: false, + }; + let env = Env::new( + sqlness_home.clone(), + ServerAddr::default(), + config.wal_config.clone(), + config.pull_version_on_need, + config.from_bins_dir.clone(), + store_config, + vec![], + ); + if let Some(overlay) = profile.overlay() { + env.activate_compat_old(overlay); + } + + // Arm immediately before starting the profile so earlier preflight failures + // cannot remove a developer-owned etcd container. + let etcd_guard = setup_etcd.then(EtcdGuard::new); + println!( + "Starting old-version {} cluster...", + config.topology.as_str() + ); + let db = match config.topology { + CompatTopology::Distributed => env.compat_start_distributed(0).await, + CompatTopology::Standalone => env.compat_start_standalone(0).await, + }; + + let mut progress = ProfileProgress::default(); + for (case_index, case) in profile.cases().iter().enumerate() { + match run_compat_phase(&db, case, config.interceptor_registry, CompatPhase::Setup).await { + Ok(()) => { + println!(" Setup: {} - OK", case.metadata.name); + progress.record_setup_success(case_index); + } + Err(error) => { + println!(" Setup: {} - FAILED: {error}", case.metadata.name); + if progress.record_setup_failure( + format!("{} (setup): {error}", case.metadata.name), + config.fail_fast, + ) { + break; + } + } + } + } + + if progress.should_transition_to_current() { + println!("Restarting cluster with new-version binary on preserved state..."); + env.activate_compat_current(); + env.compat_restart(&db, config.to_bins_dir.clone()).await; - // ---- 10. Run verify phase on the to-version cluster ---- println!("Running verify phase..."); - let mut failed = Vec::new(); - for case in &cases { - match run_compat_phase(&db, case, &interceptor_registry, CompatPhase::Verify).await { + for case_index in progress.successful_setup_indexes.clone() { + let case = &profile.cases()[case_index]; + match run_compat_phase(&db, case, config.interceptor_registry, CompatPhase::Verify) + .await + { Ok(()) => println!(" Verify: {} - PASSED", case.metadata.name), - Err(e) => { - println!(" Verify: {} - FAILED: {e}", case.metadata.name); - failed.push(case.metadata.name.clone()); - if self.fail_fast { + Err(error) => { + println!(" Verify: {} - FAILED: {error}", case.metadata.name); + if progress.record_verify_failure( + format!("{} (verify): {error}", case.metadata.name), + config.fail_fast, + ) { break; } } } } + } else if progress.successful_setup_indexes.is_empty() { + println!("Skipping current-version restart: no setup cases succeeded"); + } - // ---- 11. Stop cluster ---- - db.compat_stop(); + let cleanup = cleanup_profile(db, etcd_guard, setup_etcd, profile_state).await; + for failure in cleanup.failures { + progress.record_cleanup_failure(failure); + } + let stop_remaining_profiles = progress.should_stop_remaining_profiles(); + ProfileOutcome { + failures: progress.failures, + stop_remaining_profiles, + preserved_state: cleanup.preserved_state, + } +} - // ---- 12. Cleanup ---- - // Etcd is always cleaned up; --preserve-state only preserves sqlness_home. - if setup_etcd { - println!("Stopping etcd"); - util::stop_rm_etcd(); +struct CleanupOutcome { + failures: Vec, + preserved_state: Option, +} + +/// Owns one profile state directory until normal cleanup or unwind finalization. +struct ProfileStateGuard { + temp_dir: Option, + preserve_state: bool, +} + +impl ProfileStateGuard { + fn new(temp_dir: tempfile::TempDir, preserve_state: bool) -> Self { + Self { + temp_dir: Some(temp_dir), + preserve_state, + } + } + + fn path(&self) -> &std::path::Path { + self.temp_dir.as_ref().unwrap().path() + } + + fn finalize(mut self) -> CleanupOutcome { + let temp_dir = self.temp_dir.take().unwrap(); + let mut failures = Vec::new(); + if self.preserve_state { + let path = temp_dir.keep(); + if let Err(error) = std::fs::metadata(&path) { + failures.push(format!( + "failed to confirm preserved profile state {}: {error}", + path.display() + )); + } + return CleanupOutcome { + failures, + preserved_state: Some(path), + }; } - if !self.preserve_state { - println!("Removing state in {:?}", sqlness_home); - tokio::fs::remove_dir_all(sqlness_home) - .await - .unwrap_or_else(|e| println!("Warning: failed to clean up temp dir: {e}")); + println!("Removing state in {}", temp_dir.path().display()); + if let Err(error) = temp_dir.close() { + failures.push(format!("failed to remove profile state: {error}")); } + CleanupOutcome { + failures, + preserved_state: None, + } + } +} - // Disarm the etcd guard now that we've done normal cleanup. - if let Some(mut guard) = etcd_guard.take() { - guard.disarm(); +impl Drop for ProfileStateGuard { + fn drop(&mut self) { + let Some(temp_dir) = self.temp_dir.take() else { + return; + }; + if self.preserve_state { + let path = temp_dir.keep(); + println!( + "Preserved compat profile state after abnormal exit: {}", + path.display() + ); } + } +} - if failed.is_empty() { - println!("\n\x1b[32mAll compat tests passed!\x1b[0m"); - } else { - println!("\n\x1b[31mFailed cases: {}\x1b[0m", failed.join(", ")); - // Explicitly drop the guard before exit so it doesn't double-cleanup. - std::process::exit(1); +async fn cleanup_profile( + mut db: GreptimeDB, + mut etcd_guard: Option, + setup_etcd: bool, + profile_state: ProfileStateGuard, +) -> CleanupOutcome { + db.compat_stop(); + drop(db); + let mut failures = Vec::new(); + if setup_etcd { + println!("Stopping etcd"); + match util::stop_rm_etcd_checked() { + Ok(()) => { + if let Some(guard) = etcd_guard.as_mut() { + guard.disarm(); + } + } + Err(error) => failures.push(format!("profile etcd cleanup: {error}")), } } + // On failed checked cleanup this Drop performs one best-effort retry before + // profile state is finalized. + drop(etcd_guard); + + let mut state_outcome = profile_state.finalize(); + failures.append(&mut state_outcome.failures); + CleanupOutcome { + failures, + preserved_state: state_outcome.preserved_state, + } } /// Guard that stops/removes Docker etcd on drop (panic or early exit). @@ -546,13 +937,9 @@ impl Drop for EtcdGuard { fn drop(&mut self) { if self.active { println!("EtcdGuard: emergency etcd cleanup (panic or early exit)"); - // Best-effort: don't panic in Drop - let _ = std::process::Command::new("docker") - .args(["container", "stop", "etcd"]) - .status(); - let _ = std::process::Command::new("docker") - .args(["container", "rm", "etcd"]) - .status(); + if let Err(error) = util::stop_rm_etcd_checked() { + println!("EtcdGuard: emergency etcd cleanup failed: {error}"); + } } } } @@ -942,7 +1329,36 @@ fn simple_diff(expected: &str, actual: &str) -> String { #[cfg(test)] mod tests { - use super::trim_trailing_blank_lines; + use std::path::Path; + + use super::*; + use crate::cmd::compat_case::{CaseMetadata, OldConfigMetadata}; + + fn test_case(temp_dir: &Path, name: &str, overlay: Option<&str>) -> CompatCase { + let case_dir = temp_dir.join(name); + std::fs::create_dir_all(&case_dir).unwrap(); + if let Some(overlay) = overlay { + std::fs::write(case_dir.join("overlay.toml"), overlay).unwrap(); + } + CompatCase { + metadata: CaseMetadata { + name: name.to_string(), + reason: "test".to_string(), + introduced_by: "test".to_string(), + topologies: vec![CompatTopology::Distributed.as_str().to_string()], + from_range: vec!["*".to_string()], + to_range: vec!["*".to_string()], + features: vec!["table".to_string()], + owner: "test".to_string(), + namespace: None, + old_config: overlay.map(|_| OldConfigMetadata { + datanode: PathBuf::from("overlay.toml"), + }), + }, + dir: case_dir, + namespace: name.to_string(), + } + } #[test] fn test_trim_trailing_blank_lines_preserves_single_final_newline() { @@ -951,4 +1367,139 @@ mod tests { assert_eq!(output, "SELECT 1;\n\n+---+\n"); } + + #[test] + fn profiles_are_baseline_first_and_group_by_full_semantic_digest() { + let temp_dir = tempfile::tempdir().unwrap(); + let profiles = prepare_compat_profiles( + vec![ + test_case(temp_dir.path(), "overlay_b", Some("[x]\nb = 2\na = 1\n")), + test_case(temp_dir.path(), "baseline", None), + test_case(temp_dir.path(), "overlay_a", Some("[x]\na = 1\nb = 2\n")), + test_case(temp_dir.path(), "overlay_c", Some("value = 3\n")), + ], + &WalConfig::RaftEngine, + ) + .unwrap(); + let profiles: Vec<_> = profiles.iter().collect(); + + assert!(matches!(profiles[0], CompatProfile::Baseline { .. })); + assert_eq!(profiles[0].case_names(), ["baseline"]); + assert_eq!(profiles[1].case_names(), ["overlay_a", "overlay_b"]); + assert_eq!(profiles[1].sources().len(), 2); + let overlay_keys: Vec<_> = profiles[1..] + .iter() + .map(|profile| match profile { + CompatProfile::Overlay { overlay, .. } => *overlay.profile_key(), + CompatProfile::Baseline { .. } => unreachable!(), + }) + .collect(); + assert!(overlay_keys.windows(2).all(|keys| keys[0] < keys[1])); + } + + #[test] + fn non_fail_fast_setup_verifies_only_successful_cases() { + let mut progress = ProfileProgress::default(); + progress.record_setup_success(0); + assert!(!progress.record_setup_failure("case_b (setup): failed".to_string(), false)); + progress.record_setup_success(2); + + assert!(progress.should_transition_to_current()); + assert_eq!(progress.successful_setup_indexes, [0, 2]); + assert!(!progress.should_stop_remaining_profiles()); + } + + #[test] + fn fail_fast_setup_blocks_current_and_later_profiles() { + let mut progress = ProfileProgress::default(); + progress.record_setup_success(0); + + assert!(progress.record_setup_failure("case_b (setup): failed".to_string(), true)); + assert!(!progress.should_transition_to_current()); + assert!(progress.should_stop_remaining_profiles()); + } + + #[test] + fn zero_successful_setups_skips_current() { + let mut progress = ProfileProgress::default(); + progress.record_setup_failure("case_a (setup): failed".to_string(), false); + + assert!(!progress.should_transition_to_current()); + assert!(!progress.should_stop_remaining_profiles()); + } + + #[test] + fn verify_fail_fast_stops_later_profiles_after_cleanup() { + let mut progress = ProfileProgress::default(); + progress.record_setup_success(0); + + assert!(progress.record_verify_failure("case_a (verify): failed".to_string(), true)); + assert!(progress.should_stop_remaining_profiles()); + } + + #[test] + fn non_fail_fast_aggregates_failures_but_cleanup_failure_stops_profiles() { + let mut progress = ProfileProgress::default(); + progress.record_setup_failure("case_a (setup): failed".to_string(), false); + progress.record_setup_success(1); + progress.record_verify_failure("case_b (verify): failed".to_string(), false); + + assert!(!progress.should_stop_remaining_profiles()); + progress.record_cleanup_failure("failed to remove profile state".to_string()); + assert_eq!(progress.failures.len(), 3); + assert!(progress.should_stop_remaining_profiles()); + } + + #[test] + fn protected_path_warning_is_sorted_and_value_free() { + let warning = format_protected_path_warning( + "abcdef123456", + vec!["case_z", "case_a"], + vec!["wal.provider".to_string(), "mode".to_string()], + ); + + assert_eq!( + warning, + "Warning: datanode overlay profile abcdef123456 touches runner-owned paths [mode, wal.provider] for cases [case_a, case_z]" + ); + assert!(!warning.contains("secret-overlay-value")); + } + + #[test] + fn profile_state_guard_preserves_state_on_forced_unwind() { + let temp_dir = tempfile::tempdir().unwrap(); + let path = temp_dir.path().to_path_buf(); + + let unwind = std::panic::catch_unwind(|| { + let _state = ProfileStateGuard::new(temp_dir, true); + panic!("forced profile unwind"); + }); + + assert!(unwind.is_err()); + assert!(path.is_dir()); + std::fs::remove_dir_all(path).unwrap(); + } + + #[test] + fn profile_state_guard_removes_state_on_drop_without_preservation() { + let temp_dir = tempfile::tempdir().unwrap(); + let path = temp_dir.path().to_path_buf(); + + drop(ProfileStateGuard::new(temp_dir, false)); + + assert!(!path.exists()); + } + + #[test] + fn profile_state_guard_normal_finalization_transfers_ownership_once() { + let temp_dir = tempfile::tempdir().unwrap(); + let path = temp_dir.path().to_path_buf(); + + let cleanup = ProfileStateGuard::new(temp_dir, true).finalize(); + + assert!(cleanup.failures.is_empty()); + assert_eq!(cleanup.preserved_state.as_deref(), Some(path.as_path())); + assert!(path.is_dir()); + std::fs::remove_dir_all(path).unwrap(); + } } diff --git a/tests/runner/src/cmd/compat_case.rs b/tests/runner/src/cmd/compat_case.rs index bc9065c449..7c6dbbfa74 100644 --- a/tests/runner/src/cmd/compat_case.rs +++ b/tests/runner/src/cmd/compat_case.rs @@ -43,6 +43,17 @@ pub struct CaseMetadata { /// Must match `[a-z0-9_]+`. #[serde(default)] pub namespace: Option, + /// Optional old-stage server configuration sidecars. + #[serde(default)] + pub old_config: Option, +} + +/// Optional configuration sidecars for the old compatibility stage. +#[derive(Debug, Clone, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct OldConfigMetadata { + /// Datanode TOML sidecar, resolved relative to the compatibility case directory. + pub datanode: PathBuf, } impl CaseMetadata { @@ -53,6 +64,13 @@ impl CaseMetadata { .clone() .unwrap_or_else(|| sanitize_namespace(case_dir_name)) } + + /// Returns the old-stage datanode sidecar reference, if configured. + pub fn old_datanode_overlay(&self) -> Option<&Path> { + self.old_config + .as_ref() + .map(|config| config.datanode.as_path()) + } } /// A loaded compatibility case (metadata + file paths). @@ -469,6 +487,74 @@ mod tests { assert_eq!(sanitize_namespace("a.b-c"), "a_b_c"); } + #[test] + fn test_case_metadata_parses_old_datanode_overlay() { + let metadata: CaseMetadata = toml::from_str( + r#" + name = "case" + reason = "reason" + introduced_by = "test" + topologies = ["distributed"] + from_range = ["*"] + to_range = ["*"] + features = ["table"] + owner = "team" + + [old_config] + datanode = "old-datanode.overlay.toml" + "#, + ) + .unwrap(); + + assert_eq!( + metadata.old_datanode_overlay(), + Some(Path::new("old-datanode.overlay.toml")) + ); + } + + #[test] + fn test_case_metadata_rejects_empty_old_config() { + let error = toml::from_str::( + r#" + name = "case" + reason = "reason" + introduced_by = "test" + topologies = ["distributed"] + from_range = ["*"] + to_range = ["*"] + features = ["table"] + owner = "team" + + [old_config] + "#, + ) + .unwrap_err(); + + assert!(error.to_string().contains("missing field `datanode`")); + } + + #[test] + fn test_case_metadata_rejects_unknown_old_config_fields() { + let error = toml::from_str::( + r#" + name = "case" + reason = "reason" + introduced_by = "test" + topologies = ["distributed"] + from_range = ["*"] + to_range = ["*"] + features = ["table"] + owner = "team" + + [old_config] + unsupported = "value" + "#, + ) + .unwrap_err(); + + assert!(error.to_string().contains("unknown field `unsupported`")); + } + #[test] fn test_validate_cases_metadata_rejects_empty_required_vectors() { let case = CompatCase { @@ -482,6 +568,7 @@ mod tests { features: vec!["table".to_string()], owner: "team".to_string(), namespace: None, + old_config: None, }, dir: PathBuf::from("case"), namespace: "case".to_string(), @@ -503,6 +590,7 @@ mod tests { features: vec!["table".to_string()], owner: "test".to_string(), namespace: None, + old_config: None, }, dir: PathBuf::from("bad_constraint"), namespace: "bad_constraint".to_string(), @@ -524,6 +612,7 @@ mod tests { features: vec!["table".to_string()], owner: "test".to_string(), namespace: None, + old_config: None, }, dir: PathBuf::from("case_a"), namespace: "shared_name".to_string(), @@ -539,6 +628,7 @@ mod tests { features: vec!["table".to_string()], owner: "test".to_string(), namespace: None, + old_config: None, }, dir: PathBuf::from("case_b"), namespace: "shared_name".to_string(), @@ -563,6 +653,7 @@ mod tests { features: vec!["table".to_string()], owner: "test".to_string(), namespace: Some("shared_name".to_string()), + old_config: None, }, dir: PathBuf::from("case_a"), namespace: "shared_name".to_string(), @@ -578,6 +669,7 @@ mod tests { features: vec!["table".to_string()], owner: "test".to_string(), namespace: Some("shared_name".to_string()), + old_config: None, }, dir: PathBuf::from("case_b"), namespace: "shared_name".to_string(), @@ -599,6 +691,7 @@ mod tests { features: vec!["table".to_string()], owner: "team".to_string(), namespace: None, + old_config: None, }, dir: PathBuf::from("case"), namespace: "case".to_string(), diff --git a/tests/runner/src/cmd/datanode_overlay.rs b/tests/runner/src/cmd/datanode_overlay.rs new file mode 100644 index 0000000000..21892b9a09 --- /dev/null +++ b/tests/runner/src/cmd/datanode_overlay.rs @@ -0,0 +1,637 @@ +// Copyright 2023 Greptime 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. + +use std::fs::File; +use std::io::Read; +use std::path::{Component, Path, PathBuf}; + +use sha2::{Digest, Sha256}; +use toml::value::Table; + +use crate::env::bare::WalConfig; +/// A dotted TOML path owned by the compatibility runner. +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] +pub(crate) struct DottedPath(Vec); + +impl DottedPath { + fn new(parts: &[&str]) -> Self { + Self(parts.iter().map(|part| (*part).to_string()).collect()) + } + + fn parts(&self) -> impl Iterator { + self.0.iter().map(String::as_str) + } +} + +impl std::fmt::Display for DottedPath { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(&self.0.join(".")) + } +} + +/// Runner-owned datanode configuration paths that an overlay cannot override. +#[derive(Debug, Clone)] +pub(crate) struct DatanodeProtectionPolicy { + protected_paths: Vec, +} + +impl DatanodeProtectionPolicy { + /// Builds the policy for the runner-selected WAL configuration. + pub(crate) fn for_wal(wal: &WalConfig) -> Self { + let mut protected_paths = vec![ + DottedPath::new(&["mode"]), + DottedPath::new(&["node_id"]), + DottedPath::new(&["storage", "data_home"]), + DottedPath::new(&["meta_client_options", "metasrv_addrs"]), + DottedPath::new(&["wal", "provider"]), + ]; + protected_paths.push(match wal { + WalConfig::RaftEngine => DottedPath::new(&["wal", "dir"]), + WalConfig::Kafka { .. } => DottedPath::new(&["wal", "broker_endpoints"]), + }); + protected_paths.sort(); + + Self { protected_paths } + } +} + +/// A parsed old-stage datanode sidecar, loaded from a confined case directory. +#[derive(Debug, Clone)] +pub(crate) struct DatanodeOverlay { + source: PathBuf, + value: toml::Value, + profile_key: [u8; 32], + profile_id: String, +} + +impl DatanodeOverlay { + /// Loads and parses a datanode sidecar referenced relative to `case_dir`. + /// + /// This is intended to prevent accidental case-directory escapes; it is not + /// an adversarial concurrent-filesystem security boundary. + pub(crate) fn load(case_dir: &Path, relative_ref: &Path) -> Result { + validate_relative_reference(relative_ref)?; + + let canonical_case_dir = case_dir.canonicalize().map_err(|error| { + format!( + "Failed to canonicalize compatibility case directory {}: {error}", + case_dir.display() + ) + })?; + let requested_path = case_dir.join(relative_ref); + let canonical_target = requested_path.canonicalize().map_err(|error| { + format!( + "Failed to resolve datanode overlay {} relative to case directory {}: {error}", + relative_ref.display(), + case_dir.display() + ) + })?; + + if !canonical_target.starts_with(&canonical_case_dir) { + return Err(format!( + "Datanode overlay {} resolves outside compatibility case directory {}", + relative_ref.display(), + canonical_case_dir.display() + )); + } + + let mut file = File::open(&canonical_target).map_err(|error| { + format!( + "Failed to open datanode overlay {}: {error}", + canonical_target.display() + ) + })?; + let metadata = file.metadata().map_err(|error| { + format!( + "Failed to inspect datanode overlay {}: {error}", + canonical_target.display() + ) + })?; + if !metadata.is_file() { + return Err(format!( + "Datanode overlay {} must be a regular file", + canonical_target.display() + )); + } + + let mut content = String::new(); + file.read_to_string(&mut content).map_err(|error| { + format!( + "Failed to read datanode overlay {}: {error}", + canonical_target.display() + ) + })?; + let value: toml::Value = toml::from_str(&content).map_err(|error| { + format!( + "Failed to parse datanode overlay {}: {error}", + canonical_target.display() + ) + })?; + if !value.is_table() { + return Err(format!( + "Datanode overlay {} must have a TOML table at its root", + canonical_target.display() + )); + } + + let profile_key = semantic_profile_key(&value); + let profile_id = hex::encode(profile_key)[..12].to_string(); + + Ok(Self { + source: canonical_target, + value, + profile_key, + profile_id, + }) + } + + /// Validates runner-owned path conflicts and records overridden protected paths. + pub(crate) fn prepare( + self, + protection: &DatanodeProtectionPolicy, + ) -> Result { + let mut touched_protected_paths = Vec::new(); + for path in &protection.protected_paths { + validate_protected_ancestors(&self.value, path).map_err(|error| { + format!( + "Datanode overlay {} (profile {}): {error}", + self.source.display(), + self.profile_id + ) + })?; + if value_at_path(&self.value, path).is_some() { + touched_protected_paths.push(path.clone()); + } + } + touched_protected_paths.sort(); + + Ok(PreparedDatanodeOverlay { + source: self.source, + value: self.value, + profile_key: self.profile_key, + profile_id: self.profile_id, + protected_paths: protection.protected_paths.clone(), + touched_protected_paths, + }) + } +} + +/// A validated datanode overlay ready for application to a rendered baseline. +#[derive(Debug, Clone)] +pub(crate) struct PreparedDatanodeOverlay { + source: PathBuf, + value: toml::Value, + profile_key: [u8; 32], + profile_id: String, + protected_paths: Vec, + touched_protected_paths: Vec, +} + +impl PreparedDatanodeOverlay { + /// Returns the canonical full SHA-256 profile key used for grouping. + pub(crate) fn profile_key(&self) -> &[u8; 32] { + &self.profile_key + } + + /// Returns a truncated profile ID suitable only for diagnostics. + pub(crate) fn profile_id(&self) -> &str { + &self.profile_id + } + + /// Returns the canonical sidecar path used for diagnostics. + pub(crate) fn source(&self) -> &Path { + &self.source + } + + /// Returns protected paths declared by the sidecar, in dotted-path order. + pub(crate) fn touched_protected_paths(&self) -> &[DottedPath] { + &self.touched_protected_paths + } + + /// Merges the sidecar into a rendered baseline and restores runner-owned fields. + pub(crate) fn apply_to_rendered_baseline(&self, baseline: &str) -> Result { + let baseline_value: toml::Value = toml::from_str(baseline) + .map_err(|error| format!("Failed to parse rendered datanode baseline: {error}"))?; + if !baseline_value.is_table() { + return Err( + "Rendered datanode baseline must have a TOML table at its root".to_string(), + ); + } + + let mut merged = baseline_value.clone(); + merge_toml_values(&mut merged, &self.value); + + for path in &self.protected_paths { + restore_protected_path(&mut merged, &baseline_value, path)?; + } + + toml::to_string(&merged) + .map_err(|error| format!("Failed to serialize merged datanode configuration: {error}")) + } +} + +fn validate_relative_reference(relative_ref: &Path) -> Result<(), String> { + if relative_ref.as_os_str().is_empty() + || relative_ref + .components() + .any(|component| !matches!(component, Component::Normal(_))) + { + return Err(format!( + "Datanode overlay reference {} must be a non-empty relative path with normal components only", + relative_ref.display() + )); + } + Ok(()) +} + +fn validate_protected_ancestors(value: &toml::Value, path: &DottedPath) -> Result<(), String> { + let mut current = value; + let parts: Vec<_> = path.parts().collect(); + for (index, part) in parts.iter().enumerate().take(parts.len().saturating_sub(1)) { + let Some(table) = current.as_table() else { + return Err(format!( + "Datanode overlay makes ancestor {} of protected path {} non-table", + parts[..index].join("."), + path + )); + }; + let Some(next) = table.get(*part) else { + return Ok(()); + }; + if !next.is_table() { + return Err(format!( + "Datanode overlay makes ancestor {} of protected path {} non-table", + parts[..=index].join("."), + path + )); + } + current = next; + } + Ok(()) +} + +fn value_at_path<'a>(value: &'a toml::Value, path: &DottedPath) -> Option<&'a toml::Value> { + let mut current = value; + for part in path.parts() { + current = current.as_table()?.get(part)?; + } + Some(current) +} + +fn merge_toml_values(baseline: &mut toml::Value, overlay: &toml::Value) { + match (baseline, overlay) { + (toml::Value::Table(baseline), toml::Value::Table(overlay)) => { + for (key, overlay_value) in overlay { + match baseline.get_mut(key) { + Some(baseline_value) => merge_toml_values(baseline_value, overlay_value), + None => { + baseline.insert(key.clone(), overlay_value.clone()); + } + } + } + } + (baseline, overlay) => *baseline = overlay.clone(), + } +} + +fn restore_protected_path( + merged: &mut toml::Value, + baseline: &toml::Value, + path: &DottedPath, +) -> Result<(), String> { + let Some(merged_table) = merged.as_table_mut() else { + return Err("Merged datanode configuration must have a TOML table at its root".to_string()); + }; + let baseline_value = value_at_path(baseline, path).cloned(); + restore_path_in_table(merged_table, &path.0, baseline_value); + Ok(()) +} + +fn restore_path_in_table(table: &mut Table, parts: &[String], baseline_value: Option) { + let Some((part, remaining)) = parts.split_first() else { + return; + }; + if remaining.is_empty() { + match baseline_value { + Some(value) => { + table.insert(part.clone(), value); + } + None => { + table.remove(part); + } + } + return; + } + + let Some(value) = table.get_mut(part) else { + return; + }; + if let Some(table) = value.as_table_mut() { + restore_path_in_table(table, remaining, baseline_value); + } +} + +fn semantic_profile_key(value: &toml::Value) -> [u8; 32] { + let mut encoded = Vec::new(); + encode_semantic_value(&mut encoded, value); + Sha256::digest(encoded).into() +} + +fn encode_semantic_value(output: &mut Vec, value: &toml::Value) { + match value { + toml::Value::String(value) => encode_bytes(output, b"string", value.as_bytes()), + toml::Value::Integer(value) => { + encode_bytes(output, b"integer", &value.to_be_bytes()); + } + toml::Value::Float(value) => { + encode_bytes(output, b"float", &value.to_bits().to_be_bytes()); + } + toml::Value::Boolean(value) => { + encode_bytes(output, b"boolean", &[u8::from(*value)]); + } + toml::Value::Datetime(value) => { + encode_bytes(output, b"datetime", value.to_string().as_bytes()) + } + toml::Value::Array(values) => { + let mut payload = Vec::new(); + encode_length(&mut payload, values.len()); + for value in values { + encode_semantic_value(&mut payload, value); + } + encode_bytes(output, b"array", &payload); + } + toml::Value::Table(table) => { + let mut payload = Vec::new(); + encode_length(&mut payload, table.len()); + let mut entries: Vec<_> = table.iter().collect(); + entries.sort_unstable_by(|(left, _), (right, _)| left.as_bytes().cmp(right.as_bytes())); + for (key, value) in entries { + encode_bytes(&mut payload, b"key", key.as_bytes()); + encode_semantic_value(&mut payload, value); + } + encode_bytes(output, b"table", &payload); + } + } +} + +fn encode_bytes(output: &mut Vec, tag: &[u8], value: &[u8]) { + encode_length(output, tag.len()); + output.extend_from_slice(tag); + encode_length(output, value.len()); + output.extend_from_slice(value); +} + +fn encode_length(output: &mut Vec, length: usize) { + output.extend_from_slice(&(length as u64).to_be_bytes()); +} + +#[cfg(test)] +mod tests { + use super::*; + + fn load_overlay(content: &str) -> DatanodeOverlay { + let temp_dir = tempfile::tempdir().unwrap(); + let path = temp_dir.path().join("overlay.toml"); + std::fs::write(&path, content).unwrap(); + let overlay = DatanodeOverlay::load(temp_dir.path(), Path::new("overlay.toml")).unwrap(); + // `load` owns the parsed TOML, so the temporary source need not outlive it. + overlay + } + + fn prepared_overlay(content: &str, wal: &WalConfig) -> PreparedDatanodeOverlay { + load_overlay(content) + .prepare(&DatanodeProtectionPolicy::for_wal(wal)) + .unwrap() + } + + #[test] + fn merges_recursive_tables_and_replaces_non_tables_atomically() { + let prepared = prepared_overlay( + r#" + scalar = "replacement" + primitive_array = [3, 2, 1] + array_of_tables = [{ name = "new" }] + [nested] + new_key = true + old_key = "replacement" + "#, + &WalConfig::RaftEngine, + ); + let merged = prepared + .apply_to_rendered_baseline( + r#" + scalar = 1 + primitive_array = [1, 2] + array_of_tables = [{ name = "old" }, { name = "older" }] + untouched = "kept" + [nested] + old_key = 1 + retained = "kept" + "#, + ) + .unwrap(); + let value: toml::Value = toml::from_str(&merged).unwrap(); + + assert_eq!(value["scalar"].as_str(), Some("replacement")); + assert_eq!(value["primitive_array"].as_array().unwrap().len(), 3); + assert_eq!(value["array_of_tables"].as_array().unwrap().len(), 1); + assert_eq!(value["nested"]["old_key"].as_str(), Some("replacement")); + assert_eq!(value["nested"]["new_key"].as_bool(), Some(true)); + assert_eq!(value["nested"]["retained"].as_str(), Some("kept")); + assert_eq!(value["untouched"].as_str(), Some("kept")); + } + + #[test] + fn restores_protected_paths_and_deletes_absent_raft_wal_dir() { + let prepared = prepared_overlay( + r#" + mode = "standalone" + node_id = 99 + [storage] + data_home = "/overlay/data" + [meta_client_options] + metasrv_addrs = ["overlay:3002"] + [wal] + provider = "kafka" + dir = "/overlay/wal" + tuning = 42 + "#, + &WalConfig::RaftEngine, + ); + let merged = prepared + .apply_to_rendered_baseline( + r#" + mode = "distributed" + node_id = 1 + [storage] + data_home = "/runner/data" + [meta_client_options] + metasrv_addrs = ["runner:3002"] + [wal] + provider = "raft_engine" + "#, + ) + .unwrap(); + let value: toml::Value = toml::from_str(&merged).unwrap(); + + assert_eq!(value["mode"].as_str(), Some("distributed")); + assert_eq!(value["node_id"].as_integer(), Some(1)); + assert_eq!(value["storage"]["data_home"].as_str(), Some("/runner/data")); + assert_eq!( + value["meta_client_options"]["metasrv_addrs"][0].as_str(), + Some("runner:3002") + ); + assert_eq!(value["wal"]["provider"].as_str(), Some("raft_engine")); + assert!(value["wal"].get("dir").is_none()); + assert_eq!(value["wal"]["tuning"].as_integer(), Some(42)); + } + + #[test] + fn kafka_policy_restores_broker_endpoints() { + let prepared = prepared_overlay( + r#" + [wal] + broker_endpoints = ["overlay:9092"] + provider = "raft_engine" + "#, + &WalConfig::Kafka { + needs_kafka_cluster: false, + broker_endpoints: vec![], + }, + ); + let merged = prepared + .apply_to_rendered_baseline( + r#" + [wal] + provider = "kafka" + broker_endpoints = ["runner:9092"] + "#, + ) + .unwrap(); + let value: toml::Value = toml::from_str(&merged).unwrap(); + + assert_eq!(value["wal"]["provider"].as_str(), Some("kafka")); + assert_eq!( + value["wal"]["broker_endpoints"][0].as_str(), + Some("runner:9092") + ); + } + + #[test] + fn rejects_non_table_protected_ancestor() { + let overlay = load_overlay("wal = \"not a table\""); + let error = overlay + .prepare(&DatanodeProtectionPolicy::for_wal(&WalConfig::RaftEngine)) + .unwrap_err(); + + assert!(error.contains("ancestor wal")); + assert!(error.contains("profile")); + assert!(error.contains("wal.dir") || error.contains("wal.provider")); + } + + #[test] + fn collects_sorted_touched_protected_paths() { + let prepared = prepared_overlay( + r#" + node_id = 4 + [wal] + dir = "/overlay/wal" + provider = "kafka" + "#, + &WalConfig::RaftEngine, + ); + + let paths: Vec<_> = prepared + .touched_protected_paths() + .iter() + .map(ToString::to_string) + .collect(); + assert_eq!(paths, ["node_id", "wal.dir", "wal.provider"]); + } + + #[test] + fn semantic_identity_ignores_formatting_and_table_order() { + let first = load_overlay("[settings]\nb = 2\na = \"value\"\n"); + let second = load_overlay("# comment\n[settings]\na = \"value\"\nb = 2\n"); + + assert_eq!(first.profile_key, second.profile_key); + assert_eq!(first.profile_id, second.profile_id); + } + + #[test] + fn semantic_identity_preserves_scalar_types_and_array_order() { + let integer = load_overlay("value = 1"); + let float = load_overlay("value = 1.0"); + let first_order = load_overlay("values = [1, 2]"); + let second_order = load_overlay("values = [2, 1]"); + + assert_ne!(integer.profile_key, float.profile_key); + assert_ne!(first_order.profile_key, second_order.profile_key); + } + + #[test] + fn prepared_overlay_exposes_profile_and_source_for_runner_grouping() { + let overlay = load_overlay("value = 1"); + let expected_key = overlay.profile_key; + let expected_id = overlay.profile_id.clone(); + let prepared = overlay + .prepare(&DatanodeProtectionPolicy::for_wal(&WalConfig::RaftEngine)) + .unwrap(); + + assert_eq!(prepared.profile_key(), &expected_key); + assert_eq!(prepared.profile_id(), expected_id); + assert!(prepared.source().ends_with("overlay.toml")); + } + + #[test] + fn rejects_absolute_and_traversal_references() { + let temp_dir = tempfile::tempdir().unwrap(); + let absolute = temp_dir.path().join("overlay.toml"); + std::fs::write(&absolute, "value = 1").unwrap(); + + assert!(DatanodeOverlay::load(temp_dir.path(), &absolute).is_err()); + assert!(DatanodeOverlay::load(temp_dir.path(), Path::new("../overlay.toml")).is_err()); + assert!(DatanodeOverlay::load(temp_dir.path(), Path::new("./overlay.toml")).is_err()); + } + + #[test] + fn rejects_directories_and_parse_errors() { + let temp_dir = tempfile::tempdir().unwrap(); + std::fs::create_dir(temp_dir.path().join("directory.toml")).unwrap(); + std::fs::write(temp_dir.path().join("broken.toml"), "[broken").unwrap(); + + let directory_error = + DatanodeOverlay::load(temp_dir.path(), Path::new("directory.toml")).unwrap_err(); + assert!(directory_error.contains("regular file")); + let parse_error = + DatanodeOverlay::load(temp_dir.path(), Path::new("broken.toml")).unwrap_err(); + assert!(parse_error.contains("Failed to parse datanode overlay")); + assert!(parse_error.contains("broken.toml")); + } + + #[cfg(unix)] + #[test] + fn rejects_symlink_escape() { + use std::os::unix::fs::symlink; + + let case_dir = tempfile::tempdir().unwrap(); + let outside_dir = tempfile::tempdir().unwrap(); + let outside_file = outside_dir.path().join("outside.toml"); + std::fs::write(&outside_file, "value = 1").unwrap(); + symlink(&outside_file, case_dir.path().join("escape.toml")).unwrap(); + + let error = DatanodeOverlay::load(case_dir.path(), Path::new("escape.toml")).unwrap_err(); + assert!(error.contains("outside compatibility case directory")); + } +} diff --git a/tests/runner/src/env/bare.rs b/tests/runner/src/env/bare.rs index ed248a1fe7..8fc3855edb 100644 --- a/tests/runner/src/env/bare.rs +++ b/tests/runner/src/env/bare.rs @@ -31,6 +31,7 @@ use tokio::sync::Mutex as TokioMutex; use crate::client::MultiProtocolClient; use crate::cmd::bare::ServerAddr; use crate::cmd::compat_case::try_infer_version; +use crate::cmd::datanode_overlay::PreparedDatanodeOverlay; use crate::formatter::{ErrorFormatter, MysqlFormatter, OutputFormatter, PostgresqlFormatter}; use crate::protocol_interceptor::{MYSQL, PROTOCOL_KEY}; use crate::server_mode::{GrpcArgStyle, ServerMode}; @@ -104,6 +105,39 @@ pub struct Env { extra_args: Vec, /// Cache for the inferred gRPC argument style per `bins_dir`. grpc_arg_style_cache: Arc>>, + compat_config_stage: Arc>, +} + +/// Kills a process unless ownership has been transferred to [`GreptimeDB`]. +struct ChildGuard(Option); + +impl ChildGuard { + fn new(child: Child) -> Self { + Self(Some(child)) + } + + fn into_inner(mut self) -> Child { + self.0.take().unwrap() + } +} + +impl Drop for ChildGuard { + fn drop(&mut self) { + if let Some(child) = self.0.as_mut() { + Env::stop_server(child); + } + } +} + +/// Compatibility configuration selected for future server renders. +#[derive(Clone, Debug)] +pub(crate) enum CompatConfigStage { + /// Ordinary and baseline compatibility renders use the template unchanged. + Baseline, + /// Old compatibility renders apply this datanode overlay. + Old(Arc), + /// Current compatibility renders use the template unchanged. + Current, } #[async_trait] @@ -157,9 +191,25 @@ impl Env { store_config, extra_args, grpc_arg_style_cache: Arc::new(Mutex::new(HashMap::new())), + compat_config_stage: Arc::new(Mutex::new(CompatConfigStage::Baseline)), } } + /// Selects the old-stage overlay for subsequent compatibility renders. + pub(crate) fn activate_compat_old(&self, overlay: Arc) { + *self.compat_config_stage.lock().unwrap() = CompatConfigStage::Old(overlay); + } + + /// Selects clean current-stage rendering for subsequent compatibility renders. + pub(crate) fn activate_compat_current(&self) { + *self.compat_config_stage.lock().unwrap() = CompatConfigStage::Current; + } + + /// Takes a cheap compatibility-stage snapshot before rendering or spawning. + pub(crate) fn compat_config_stage(&self) -> CompatConfigStage { + self.compat_config_stage.lock().unwrap().clone() + } + async fn start_standalone(&self, id: usize) -> GreptimeDB { println!("Starting standalone instance id: {id}"); @@ -176,7 +226,8 @@ impl Env { let server_process = self.start_server(server_mode, &db_ctx, id, true).await; let mut greptimedb = self.connect_db(&server_addr, id).await; - greptimedb.server_processes = Some(Arc::new(Mutex::new(vec![server_process]))); + greptimedb.server_processes = + Some(Arc::new(Mutex::new(vec![server_process.into_inner()]))); greptimedb.is_standalone = true; greptimedb.ctx = db_ctx; @@ -237,10 +288,12 @@ impl Env { let mut greptimedb = self.connect_db(&server_addr, id).await; - greptimedb.metasrv_process = Some(meta_server).into(); - greptimedb.server_processes = Some(Arc::new(Mutex::new(datanodes))); - greptimedb.frontend_process = Some(frontend).into(); - greptimedb.flownode_process = Some(flownode).into(); + greptimedb.metasrv_process = Some(meta_server.into_inner()).into(); + greptimedb.server_processes = Some(Arc::new(Mutex::new( + datanodes.into_iter().map(ChildGuard::into_inner).collect(), + ))); + greptimedb.frontend_process = Some(frontend.into_inner()).into(); + greptimedb.flownode_process = Some(flownode.into_inner()).into(); greptimedb.is_standalone = false; greptimedb.ctx = db_ctx; @@ -311,7 +364,7 @@ impl Env { db_ctx: &GreptimeDBContext, id: usize, truncate_log: bool, - ) -> Child { + ) -> ChildGuard { let bins_dir = self.bins_dir.lock().unwrap().clone().expect( "GreptimeDB binary is not available. Please pass in the path to the directory that contains the pre-built GreptimeDB binary. Or you may call `self.build_db()` beforehand.", ); @@ -327,7 +380,7 @@ impl Env { id: usize, truncate_log: bool, bins_dir: PathBuf, - ) -> Child { + ) -> ChildGuard { let log_file_name = match mode { ServerMode::Datanode { node_id, .. } => { db_ctx.incr_datanode_id(); @@ -351,7 +404,15 @@ impl Env { .unwrap(); let arg_style = self.infer_grpc_arg_style(&bins_dir); - let args = mode.get_args(&self.sqlness_home, self, db_ctx, id, arg_style); + let compat_stage = self.compat_config_stage(); + let args = mode.get_args( + &self.sqlness_home, + self, + db_ctx, + id, + arg_style, + &compat_stage, + ); let check_ip_addrs = mode.check_addrs(); for check_ip_addr in &check_ip_addrs { @@ -369,7 +430,7 @@ impl Env { .canonicalize() .expect("Failed to canonicalize bins_dir"); - let mut process = Command::new(abs_bins_dir.join(program)) + let process = Command::new(abs_bins_dir.join(program)) .current_dir(bins_dir.clone()) .env("TZ", "UTC") .args(args) @@ -382,10 +443,10 @@ impl Env { bins_dir.join(program) ); }); + let process = ChildGuard::new(process); for check_ip_addr in &check_ip_addrs { if !util::check_port(check_ip_addr.parse().unwrap(), Duration::from_secs(30)).await { - Env::stop_server(&mut process); panic!( "{} doesn't up in 30 seconds, check {} for more details.", mode.name(), @@ -454,6 +515,7 @@ impl Env { vec![new_server_process] } else { db.ctx.reset_datanode_id(); + let mut new_metasrv = None; if is_full_restart { let metasrv_mode = db .ctx @@ -469,10 +531,7 @@ impl Env { bins_dir.clone(), ) .await; - db.metasrv_process - .lock() - .expect("lock poisoned") - .replace(metasrv); + new_metasrv = Some(metasrv); // wait for metasrv to start // since it seems older version of db might take longer to complete election @@ -498,6 +557,7 @@ impl Env { processes.push(new_server_process); } + let mut new_frontend = None; if is_full_restart { let frontend_mode = db .ctx @@ -514,10 +574,6 @@ impl Env { bins_dir.clone(), ) .await; - db.frontend_process - .lock() - .expect("lock poisoned") - .replace(frontend); // Reconnect protocol clients to the new frontend process // so that MySQL/Postgres queries use the restarted frontend, @@ -529,9 +585,11 @@ impl Env { client .reconnect_pg_client(server_addr.pg_server_addr.as_ref().unwrap()) .await; + new_frontend = Some(frontend); } // Restart flownode. + let mut new_flownode = None; if let Some(flownode_mode) = db.ctx.get_server_mode(SERVER_MODE_FLOWNODE_IDX).cloned() { let flownode = self .start_server_with_bins_dir( @@ -542,10 +600,20 @@ impl Env { bins_dir.clone(), ) .await; - db.flownode_process - .lock() - .expect("lock poisoned") - .replace(flownode); + new_flownode = Some(flownode); + } + + if let Some(metasrv) = new_metasrv { + let mut metasrv_process = db.metasrv_process.lock().expect("lock poisoned"); + metasrv_process.replace(metasrv.into_inner()); + } + if let Some(frontend) = new_frontend { + let mut frontend_process = db.frontend_process.lock().expect("lock poisoned"); + frontend_process.replace(frontend.into_inner()); + } + if let Some(flownode) = new_flownode { + let mut flownode_process = db.flownode_process.lock().expect("lock poisoned"); + flownode_process.replace(flownode.into_inner()); } processes @@ -553,7 +621,10 @@ impl Env { if let Some(server_processes) = db.server_processes.clone() { let mut server_processes = server_processes.lock().unwrap(); - *server_processes = new_server_processes; + *server_processes = new_server_processes + .into_iter() + .map(ChildGuard::into_inner) + .collect(); } } @@ -969,3 +1040,100 @@ impl GreptimeDBContext { self.server_modes.get(idx) } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::cmd::bare::ServerAddr; + use crate::cmd::datanode_overlay::{DatanodeOverlay, DatanodeProtectionPolicy}; + + fn test_env(temp_dir: &Path) -> Env { + Env::new( + temp_dir.to_path_buf(), + ServerAddr::default(), + WalConfig::RaftEngine, + false, + None, + StoreConfig { + store_addrs: vec![], + setup_etcd: false, + setup_pg: None, + setup_mysql: None, + enable_flat_format: false, + enable_gc: false, + }, + vec![], + ) + } + + #[test] + fn compat_config_stage_is_shared_by_env_clones_and_transitions_cleanly() { + let temp_dir = tempfile::tempdir().unwrap(); + std::fs::write(temp_dir.path().join("overlay.toml"), "value = 1").unwrap(); + let overlay = DatanodeOverlay::load(temp_dir.path(), Path::new("overlay.toml")) + .unwrap() + .prepare(&DatanodeProtectionPolicy::for_wal(&WalConfig::RaftEngine)) + .unwrap(); + let env = test_env(temp_dir.path()); + let clone = env.clone(); + + assert!(matches!( + clone.compat_config_stage(), + CompatConfigStage::Baseline + )); + env.activate_compat_old(Arc::new(overlay)); + assert!(matches!( + clone.compat_config_stage(), + CompatConfigStage::Old(_) + )); + clone.activate_compat_current(); + assert!(matches!( + env.compat_config_stage(), + CompatConfigStage::Current + )); + } + + #[cfg(unix)] + #[test] + fn child_guard_kills_untransferred_processes() { + let child = std::process::Command::new("sleep") + .arg("60") + .spawn() + .unwrap(); + let pid = child.id().to_string(); + + drop(ChildGuard::new(child)); + + assert!( + !std::process::Command::new("kill") + .args(["-0", &pid]) + .status() + .unwrap() + .success() + ); + } + + #[cfg(unix)] + #[test] + fn child_guard_kills_process_on_forced_unwind() { + let child = std::process::Command::new("sleep") + .arg("60") + .spawn() + .unwrap(); + let pid = child.id().to_string(); + + let unwind = std::panic::catch_unwind(|| { + let _guard = ChildGuard::new(child); + panic!("forced startup unwind"); + }); + + assert!(unwind.is_err()); + assert!( + !std::process::Command::new("kill") + .args(["-0", &pid]) + .status() + .unwrap() + .success() + ); + } +} diff --git a/tests/runner/src/server_mode.rs b/tests/runner/src/server_mode.rs index 43c2d0cca8..1d59403af1 100644 --- a/tests/runner/src/server_mode.rs +++ b/tests/runner/src/server_mode.rs @@ -21,7 +21,7 @@ use tinytemplate::TinyTemplate; use crate::cmd::bare::ServerAddr; use crate::cmd::compat_case::Version; -use crate::env::bare::{Env, GreptimeDBContext, ServiceProvider}; +use crate::env::bare::{CompatConfigStage, Env, GreptimeDBContext, ServiceProvider}; use crate::util; const DEFAULT_LOG_LEVEL: &str = "--log-level=debug,hyper=warn,tower=warn,datafusion=warn,reqwest=warn,sqlparser=warn,h2=info,opendal=info"; @@ -305,6 +305,7 @@ impl ServerMode { sqlness_home: &Path, db_ctx: &GreptimeDBContext, id: usize, + compat_stage: &CompatConfigStage, ) -> String { let mut tt = TinyTemplate::new(); @@ -365,9 +366,34 @@ impl ServerMode { }; let rendered = tt.render(self.name(), &ctx).unwrap(); + let rendered = match (self, compat_stage) { + (ServerMode::Datanode { .. }, CompatConfigStage::Old(overlay)) => overlay + .apply_to_rendered_baseline(&rendered) + .unwrap_or_else(|error| { + panic!( + "Failed to apply old datanode overlay {}: {error}", + overlay.source().display() + ) + }), + _ => rendered, + }; + let stage_suffix = match compat_stage { + CompatConfigStage::Baseline => "baseline", + CompatConfigStage::Old(_) => "old", + CompatConfigStage::Current => "current", + }; let conf_file = data_home - .join(format!("{}-{}-{}.toml", self.name(), id, db_ctx.time())) + .join(if matches!(self, ServerMode::Datanode { .. }) { + format!( + "{}-{}-{}-{stage_suffix}.toml", + self.name(), + id, + db_ctx.time() + ) + } else { + format!("{}-{}-{}.toml", self.name(), id, db_ctx.time()) + }) .display() .to_string(); println!( @@ -387,6 +413,7 @@ impl ServerMode { db_ctx: &GreptimeDBContext, id: usize, arg_style: GrpcArgStyle, + compat_stage: &CompatConfigStage, ) -> Vec { let mut args = env .extra_args() @@ -410,7 +437,7 @@ impl ServerMode { id ), "-c".to_string(), - self.generate_config_file(sqlness_home, db_ctx, id), + self.generate_config_file(sqlness_home, db_ctx, id, compat_stage), format!("--http-addr={http_addr}"), format!("{}={rpc_bind_addr}", arg_style.bind_addr_arg()), format!("--mysql-addr={mysql_addr}"), @@ -439,7 +466,7 @@ impl ServerMode { id ), "-c".to_string(), - self.generate_config_file(sqlness_home, db_ctx, id), + self.generate_config_file(sqlness_home, db_ctx, id, compat_stage), ]); } ServerMode::Metasrv { @@ -461,7 +488,7 @@ impl ServerMode { id ), "-c".to_string(), - self.generate_config_file(sqlness_home, db_ctx, id), + self.generate_config_file(sqlness_home, db_ctx, id, compat_stage), ]); if matches!( @@ -540,7 +567,7 @@ impl ServerMode { format!("--log-dir={}/logs", data_home.display()), format!("--node-id={node_id}"), "-c".to_string(), - self.generate_config_file(sqlness_home, db_ctx, id), + self.generate_config_file(sqlness_home, db_ctx, id, compat_stage), format!("--metasrv-addrs={metasrv_addr}"), ]); } @@ -573,8 +600,10 @@ impl ServerMode { #[cfg(test)] mod tests { use std::path::PathBuf; + use std::sync::Arc; use super::*; + use crate::cmd::datanode_overlay::{DatanodeOverlay, DatanodeProtectionPolicy}; use crate::env::bare::{StoreConfig, WalConfig}; fn test_env(sqlness_home: &Path) -> (Env, GreptimeDBContext) { @@ -600,6 +629,19 @@ mod tests { (env, db_ctx) } + fn old_stage(temp_dir: &Path) -> CompatConfigStage { + std::fs::write( + temp_dir.join("old-datanode.toml"), + "old_only = \"applied\"\nmode = \"standalone\"\n", + ) + .unwrap(); + let overlay = DatanodeOverlay::load(temp_dir, Path::new("old-datanode.toml")) + .unwrap() + .prepare(&DatanodeProtectionPolicy::for_wal(&WalConfig::RaftEngine)) + .unwrap(); + CompatConfigStage::Old(Arc::new(overlay)) + } + fn has_arg(args: &[String], name: &str) -> bool { let prefix = format!("{name}="); args.iter() @@ -641,7 +683,14 @@ mod tests { postgres_addr: "127.0.0.1:4003".to_string(), }; assert_uses_style( - &standalone.get_args(temp_dir, env, db_ctx, 0, style), + &standalone.get_args( + temp_dir, + env, + db_ctx, + 0, + style, + &CompatConfigStage::Baseline, + ), style, false, ); @@ -654,7 +703,14 @@ mod tests { metasrv_addr: "127.0.0.1:4001".to_string(), }; assert_uses_style( - &frontend.get_args(temp_dir, env, db_ctx, 0, style), + &frontend.get_args( + temp_dir, + env, + db_ctx, + 0, + style, + &CompatConfigStage::Baseline, + ), style, true, ); @@ -665,7 +721,14 @@ mod tests { http_addr: "127.0.0.1:4200".to_string(), }; assert_uses_style( - &metasrv.get_args(temp_dir, env, db_ctx, 0, style), + &metasrv.get_args( + temp_dir, + env, + db_ctx, + 0, + style, + &CompatConfigStage::Baseline, + ), style, true, ); @@ -678,7 +741,14 @@ mod tests { node_id: 0, }; assert_uses_style( - &datanode.get_args(temp_dir, env, db_ctx, 0, style), + &datanode.get_args( + temp_dir, + env, + db_ctx, + 0, + style, + &CompatConfigStage::Baseline, + ), style, true, ); @@ -691,7 +761,14 @@ mod tests { node_id: 0, }; assert_uses_style( - &flownode.get_args(temp_dir, env, db_ctx, 0, style), + &flownode.get_args( + temp_dir, + env, + db_ctx, + 0, + style, + &CompatConfigStage::Baseline, + ), style, true, ); @@ -759,14 +836,79 @@ mod tests { node_id: 0, }; - let metasrv_config = - std::fs::read_to_string(metasrv.generate_config_file(temp_dir.path(), &db_ctx, 0)) - .unwrap(); - let datanode_config = - std::fs::read_to_string(datanode.generate_config_file(temp_dir.path(), &db_ctx, 0)) - .unwrap(); + let metasrv_config = std::fs::read_to_string(metasrv.generate_config_file( + temp_dir.path(), + &db_ctx, + 0, + &CompatConfigStage::Baseline, + )) + .unwrap(); + let datanode_config = std::fs::read_to_string(datanode.generate_config_file( + temp_dir.path(), + &db_ctx, + 0, + &CompatConfigStage::Baseline, + )) + .unwrap(); assert!(metasrv_config.contains("[gc]\nenable = true")); assert!(datanode_config.contains("[region_engine.mito.gc]\nenable = true")); } + + #[test] + fn test_datanode_rendering_applies_only_old_stage_with_distinct_filenames() { + let temp_dir = tempfile::tempdir().unwrap(); + let (_, db_ctx) = test_env(temp_dir.path()); + let datanode = ServerMode::Datanode { + rpc_bind_addr: "127.0.0.1:4301".to_string(), + rpc_server_addr: "127.0.0.1:4301".to_string(), + http_addr: "127.0.0.1:4300".to_string(), + metasrv_addr: "127.0.0.1:4201".to_string(), + node_id: 0, + }; + let baseline_path = datanode.generate_config_file( + temp_dir.path(), + &db_ctx, + 0, + &CompatConfigStage::Baseline, + ); + let old_path = + datanode.generate_config_file(temp_dir.path(), &db_ctx, 0, &old_stage(temp_dir.path())); + let current_path = + datanode.generate_config_file(temp_dir.path(), &db_ctx, 0, &CompatConfigStage::Current); + + let baseline = std::fs::read_to_string(&baseline_path).unwrap(); + let old = std::fs::read_to_string(&old_path).unwrap(); + let current = std::fs::read_to_string(¤t_path).unwrap(); + assert!(baseline_path.ends_with("-baseline.toml")); + assert!(old_path.ends_with("-old.toml")); + assert!(current_path.ends_with("-current.toml")); + assert!(!baseline.contains("old_only")); + assert!(old.contains("old_only = \"applied\"")); + assert!(!current.contains("old_only")); + assert_eq!( + toml::from_str::(&old).unwrap()["mode"].as_str(), + toml::from_str::(&baseline).unwrap()["mode"].as_str() + ); + } + + #[test] + fn test_non_datanode_ignores_old_overlay() { + let temp_dir = tempfile::tempdir().unwrap(); + let (_, db_ctx) = test_env(temp_dir.path()); + let metasrv = ServerMode::Metasrv { + rpc_bind_addr: "127.0.0.1:4201".to_string(), + rpc_server_addr: "127.0.0.1:4201".to_string(), + http_addr: "127.0.0.1:4200".to_string(), + }; + let config_path = + metasrv.generate_config_file(temp_dir.path(), &db_ctx, 0, &old_stage(temp_dir.path())); + + assert!(!config_path.ends_with("-old.toml")); + assert!( + !std::fs::read_to_string(config_path) + .unwrap() + .contains("old_only") + ); + } } diff --git a/tests/runner/src/util.rs b/tests/runner/src/util.rs index 6c00c8a9a6..1be4183b49 100644 --- a/tests/runner/src/util.rs +++ b/tests/runner/src/util.rs @@ -285,25 +285,55 @@ pub fn setup_etcd(client_ports: Vec, peer_port: Option, etcd_version: } } -/// Stop and remove the etcd container -pub fn stop_rm_etcd() { +/// Stop and remove the etcd container, failing if it cannot be confirmed absent. +pub fn stop_rm_etcd_checked() -> Result<(), String> { let status = std::process::Command::new("docker") - .args(["container", "stop", "etcd"]) - .status(); - if status.is_err() { - panic!("Failed to stop etcd: {:?}", status); - } else { - println!("Stopped etcd"); - } - // rm the container - let status = std::process::Command::new("docker") - .args(["container", "rm", "etcd"]) - .status(); - if status.is_err() { - panic!("Failed to remove etcd container: {:?}", status); - } else { + .args(["container", "rm", "--force", "etcd"]) + .status() + .map_err(|error| format!("Failed to run Docker while removing etcd: {error}"))?; + if status.success() { println!("Removed etcd container"); + return Ok(()); } + + let listed = std::process::Command::new("docker") + .args([ + "container", + "ls", + "--all", + "--filter", + "name=^/etcd$", + "--format", + "{{.ID}}", + ]) + .output() + .map_err(|error| { + format!( + "Docker failed to remove etcd ({status}) and could not verify its absence: {error}" + ) + })?; + if !listed.status.success() { + return Err(format!( + "Docker failed to remove etcd ({status}) and listing containers failed: {}", + String::from_utf8_lossy(&listed.stderr).trim() + )); + } + if String::from_utf8_lossy(&listed.stdout).trim().is_empty() { + println!("Etcd container is already absent"); + return Ok(()); + } + + Err(format!( + "Docker failed to remove etcd container (status {status}); the container still exists" + )) +} + +/// Stop and remove the etcd container. +/// +/// Legacy callers retain panic-on-cleanup-failure behavior. Compatibility +/// profiles use [`stop_rm_etcd_checked`] to report the failure structurally. +pub fn stop_rm_etcd() { + stop_rm_etcd_checked().unwrap_or_else(|error| panic!("{error}")); } /// Set up a PostgreSQL server in docker.