diff --git a/Cargo.lock b/Cargo.lock index 385591d..9dbccbd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -111,9 +111,9 @@ dependencies = [ [[package]] name = "anyhow" -version = "1.0.101" +version = "1.0.103" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5f0e0fee31ef5ed1ba1316088939cea399010ed7731dba877ed44aeb407a75ea" +checksum = "2a4385e2e34eb35d6b3efe798b9eb88096925d87726c0798709bf56d9ed84af3" [[package]] name = "arc-swap" @@ -660,9 +660,9 @@ dependencies = [ [[package]] name = "crossbeam-epoch" -version = "0.9.18" +version = "0.9.20" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e" +checksum = "2d6914041f254d6e9176c01941b21115dcfb7089e55135a35411081bd106ef3f" dependencies = [ "crossbeam-utils", ] @@ -1700,10 +1700,12 @@ name = "kvstore-examples" version = "0.1.0" dependencies = [ "anyhow", + "async-trait", "kvstore-node", "kvstore-runtime-ext", "kvstore-runtime-workloads", "serde", + "testing-framework-app", "testing-framework-core", "testing-framework-runner-compose", "testing-framework-runner-k8s", @@ -1735,6 +1737,7 @@ dependencies = [ "async-trait", "kvstore-node", "serde", + "testing-framework-app", "testing-framework-core", "testing-framework-runner-compose", "testing-framework-runner-k8s", @@ -1749,6 +1752,7 @@ dependencies = [ "kvstore-node", "kvstore-runtime-ext", "serde", + "testing-framework-app", "testing-framework-core", "tokio", "tracing", @@ -1920,6 +1924,23 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "multi-app-examples" +version = "0.1.0" +dependencies = [ + "anyhow", + "async-trait", + "kvstore-runtime-ext", + "openraft-kv-runtime-ext", + "openraft-kv-runtime-workloads", + "serde", + "testing-framework-app", + "testing-framework-core", + "tokio", + "tracing", + "tracing-subscriber", +] + [[package]] name = "native-tls" version = "0.2.14" @@ -2085,9 +2106,11 @@ name = "openraft-kv-examples" version = "0.1.0" dependencies = [ "anyhow", + "async-trait", "openraft-kv-node", "openraft-kv-runtime-ext", "openraft-kv-runtime-workloads", + "testing-framework-app", "testing-framework-core", "testing-framework-runner-k8s", "tokio", @@ -2120,6 +2143,7 @@ dependencies = [ "async-trait", "openraft-kv-node", "reqwest", + "testing-framework-app", "testing-framework-core", "testing-framework-runner-compose", "testing-framework-runner-k8s", @@ -2134,6 +2158,7 @@ dependencies = [ "async-trait", "openraft-kv-node", "openraft-kv-runtime-ext", + "testing-framework-app", "testing-framework-core", "thiserror 2.0.18", "tokio", @@ -3602,6 +3627,17 @@ dependencies = [ "unicode-segmentation", ] +[[package]] +name = "testing-framework-app" +version = "0.1.0" +dependencies = [ + "async-trait", + "testing-framework-core", + "testing-framework-runner-local", + "thiserror 2.0.18", + "tokio", +] + [[package]] name = "testing-framework-core" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index 661629b..05d8355 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -12,6 +12,7 @@ members = [ "examples/metrics_counter/metrics-counter-node", "examples/metrics_counter/testing/integration", "examples/metrics_counter/testing/workloads", + "examples/multi_app/examples", "examples/nats/examples", "examples/nats/testing/integration", "examples/nats/testing/workloads", @@ -30,6 +31,7 @@ members = [ "examples/redis_streams/examples", "examples/redis_streams/testing/integration", "examples/redis_streams/testing/workloads", + "testing-framework/app", "testing-framework/core", "testing-framework/deployers/compose", "testing-framework/deployers/k8s", @@ -58,6 +60,7 @@ all = "allow" cfgsync-adapter = { default-features = false, path = "cfgsync/adapter" } cfgsync-artifacts = { default-features = false, path = "cfgsync/artifacts" } cfgsync-core = { default-features = false, path = "cfgsync/core" } +testing-framework-app = { default-features = false, path = "testing-framework/app" } testing-framework-core = { default-features = false, path = "testing-framework/core" } testing-framework-runner-compose = { default-features = false, path = "testing-framework/deployers/compose" } testing-framework-runner-k8s = { default-features = false, path = "testing-framework/deployers/k8s" } diff --git a/examples/README.md b/examples/README.md new file mode 100644 index 0000000..c751212 --- /dev/null +++ b/examples/README.md @@ -0,0 +1,25 @@ +# Examples + +Use the app layer first when the test is about an application or composed +system interface. + +Canonical app-layer examples: + +- `kvstore_app_host_convergence`: one local app cluster exposed through + `AppHost` +- `openraft_kv_app_host_smoke`: one richer local app cluster exposed through + `AppHost` +- `multi_app_typed_stack_smoke`: composed app stack exposing typed child and + stack handles + +The older direct `ScenarioBuilder` examples are still useful for +backend-specific coverage: + +- local uniform cluster behavior +- compose runner behavior +- k8s runner behavior +- manual cluster control + +Do not use those older examples as the pattern for new composed systems. New +multi-app tests should define an `AppDeployment`, expose typed handles, and run +through `AppHost::scenario().with_app(...)`. diff --git a/examples/kvstore/examples/Cargo.toml b/examples/kvstore/examples/Cargo.toml index 79bc82b..3580d0f 100644 --- a/examples/kvstore/examples/Cargo.toml +++ b/examples/kvstore/examples/Cargo.toml @@ -4,6 +4,10 @@ license.workspace = true name = "kvstore-examples" version.workspace = true +[[bin]] +name = "kvstore_app_host_convergence" +path = "src/bin/app_host_convergence.rs" + [[bin]] name = "kvstore_basic_convergence" path = "src/bin/basic_convergence.rs" @@ -21,9 +25,11 @@ name = "kvstore_k8s_manual_convergence" path = "src/bin/k8s_manual_convergence.rs" [dependencies] +async-trait = { workspace = true } kvstore-node = { path = "../kvstore-node" } kvstore-runtime-ext = { path = "../testing/integration" } kvstore-runtime-workloads = { path = "../testing/workloads" } +testing-framework-app = { workspace = true } testing-framework-core = { workspace = true } testing-framework-runner-compose = { workspace = true } testing-framework-runner-k8s = { workspace = true } diff --git a/examples/kvstore/examples/src/bin/app_host_convergence.rs b/examples/kvstore/examples/src/bin/app_host_convergence.rs new file mode 100644 index 0000000..c454956 --- /dev/null +++ b/examples/kvstore/examples/src/bin/app_host_convergence.rs @@ -0,0 +1,117 @@ +use std::time::Duration; + +use async_trait::async_trait; +use kvstore_runtime_ext::{KvEnv, KvLocalApp}; +use serde::{Deserialize, Serialize}; +use testing_framework_app::{ + AppHost, AppHostEnv, AppHostLocalDeployer, AppRunContextExt, AppScenarioBuilderExt, + LocalAppCluster, +}; +use testing_framework_core::scenario::{Deployer, DynError, RunContext, Workload}; +use tracing::info; + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + tracing_subscriber::fmt() + .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) + .init(); + + let mut scenario = AppHost::scenario() + .with_app(KvLocalApp::nodes(3)) + .with_run_duration(Duration::from_secs(5)) + .with_workload(KvAppHostConvergence::new(3)) + .build()?; + + let deployer = AppHostLocalDeployer::default(); + let runner = deployer.deploy(&scenario).await?; + runner.run(&mut scenario).await?; + + Ok(()) +} + +#[derive(Clone)] +struct KvAppHostConvergence { + expected_nodes: usize, +} + +impl KvAppHostConvergence { + const fn new(expected_nodes: usize) -> Self { + Self { expected_nodes } + } +} + +#[async_trait] +impl Workload for KvAppHostConvergence { + fn name(&self) -> &str { + "kv_app_host_convergence" + } + + async fn start(&self, ctx: &RunContext) -> Result<(), DynError> { + let cluster = ctx.require_app::>()?; + + ensure_cluster_shape(&cluster, self.expected_nodes)?; + put_value(&cluster, "before-restart").await?; + cluster.restart_node("node-0").await?; + cluster.wait_node_ready("node-0").await?; + put_value(&cluster, "after-restart").await?; + + info!( + nodes = self.expected_nodes, + "kv app host cluster exposes app-local node control and clients" + ); + + Ok(()) + } +} + +fn ensure_cluster_shape( + cluster: &LocalAppCluster, + expected_nodes: usize, +) -> Result<(), DynError> { + if cluster.node_count() != expected_nodes || cluster.clients().len() != expected_nodes { + return Err(format!("kv app host expected {expected_nodes} nodes").into()); + } + + if cluster.node_client("node-0").is_none() { + return Err("kv app host cannot access node-0 client".into()); + } + + if cluster.node_pid("node-0").is_none() { + return Err("kv app host cannot access node-0 process id".into()); + } + + Ok(()) +} + +async fn put_value(cluster: &LocalAppCluster, value: &str) -> Result<(), DynError> { + let Some(client) = cluster.first_client() else { + return Err("kv app host has no clients".into()); + }; + + let response: KvPutResponse = client + .put( + "/kv/app-host-smoke", + &KvPutRequest { + value: value.to_owned(), + expected_version: None, + }, + ) + .await?; + + if !response.applied { + return Err(format!("kv app host write was rejected for value {value}").into()); + } + + Ok(()) +} + +#[derive(Serialize)] +struct KvPutRequest { + value: String, + expected_version: Option, +} + +#[derive(Deserialize)] +struct KvPutResponse { + applied: bool, +} diff --git a/examples/kvstore/examples/src/bin/basic_convergence.rs b/examples/kvstore/examples/src/bin/basic_convergence.rs index 6ebc130..e9f8041 100644 --- a/examples/kvstore/examples/src/bin/basic_convergence.rs +++ b/examples/kvstore/examples/src/bin/basic_convergence.rs @@ -2,7 +2,8 @@ use std::time::Duration; use kvstore_runtime_ext::KvLocalDeployer; use kvstore_runtime_workloads::{ - KvBuilderExt, KvConverges, KvScenarioBuilder, KvTopology, KvWriteWorkload, + KvBuilderExt, KvClusterAccessible, KvConverges, KvExistingClusterApp, KvScenarioBuilder, + KvWriteWorkload, }; use testing_framework_core::scenario::Deployer; @@ -12,8 +13,9 @@ async fn main() -> anyhow::Result<()> { .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) .init(); - let mut scenario = KvScenarioBuilder::deployment_with(|_| KvTopology::new(3)) + let mut scenario = KvScenarioBuilder::with_existing_kvstore_app(KvExistingClusterApp::nodes(3)) .with_run_duration(Duration::from_secs(30)) + .with_workload(KvClusterAccessible::new(3)) .with_workload( KvWriteWorkload::new() .operations(300) diff --git a/examples/kvstore/testing/integration/Cargo.toml b/examples/kvstore/testing/integration/Cargo.toml index 9773cac..d555f8b 100644 --- a/examples/kvstore/testing/integration/Cargo.toml +++ b/examples/kvstore/testing/integration/Cargo.toml @@ -5,6 +5,7 @@ name = "kvstore-runtime-ext" version.workspace = true [dependencies] +testing-framework-app = { workspace = true } testing-framework-core = { workspace = true } testing-framework-runner-compose = { workspace = true } testing-framework-runner-k8s = { workspace = true } diff --git a/examples/kvstore/testing/integration/src/app.rs b/examples/kvstore/testing/integration/src/app.rs index bd0dffe..ebbd04e 100644 --- a/examples/kvstore/testing/integration/src/app.rs +++ b/examples/kvstore/testing/integration/src/app.rs @@ -3,9 +3,13 @@ use std::io::Error; use async_trait::async_trait; use kvstore_node::KvHttpClient; use serde::{Deserialize, Serialize}; -use testing_framework_core::scenario::{ - Application, ClusterNodeConfigApplication, ClusterNodeView, ClusterPeerView, DynError, - NodeAccess, serialize_cluster_yaml_config, +use testing_framework_app::{AppDeployment, AppHostEnv, DeployContext, LocalAppCluster}; +use testing_framework_core::{ + scenario::{ + Application, ClusterNodeConfigApplication, ClusterNodeView, ClusterPeerView, DynError, + NodeAccess, NodeClients, serialize_cluster_yaml_config, + }, + topology::DeploymentDescriptor, }; pub type KvTopology = testing_framework_core::topology::ClusterTopology; @@ -40,6 +44,102 @@ impl Application for KvEnv { } } +#[derive(Clone)] +pub struct KvStoreCluster { + deployment: KvTopology, + node_clients: NodeClients, +} + +impl KvStoreCluster { + #[must_use] + pub const fn new(deployment: KvTopology, node_clients: NodeClients) -> Self { + Self { + deployment, + node_clients, + } + } + + #[must_use] + pub fn topology(&self) -> &KvTopology { + &self.deployment + } + + #[must_use] + pub fn node_count(&self) -> usize { + self.deployment.node_count() + } + + #[must_use] + pub fn clients(&self) -> Vec { + self.node_clients.snapshot() + } + + #[must_use] + pub fn first_client(&self) -> Option { + self.node_clients + .with_clients(|clients| clients.first().cloned()) + } +} + +#[derive(Clone)] +pub struct KvExistingClusterApp { + topology: KvTopology, +} + +impl KvExistingClusterApp { + #[must_use] + pub fn nodes(nodes: usize) -> Self { + Self { + topology: KvTopology::new(nodes), + } + } + + #[must_use] + pub fn topology(&self) -> KvTopology { + self.topology.clone() + } +} + +#[async_trait] +impl AppDeployment for KvExistingClusterApp { + type Handle = KvStoreCluster; + + async fn deploy(self, ctx: &mut DeployContext) -> Result { + Ok(KvStoreCluster::new( + ctx.deployment().clone(), + ctx.node_clients().clone(), + )) + } +} + +#[derive(Clone)] +pub struct KvLocalApp { + deployment: KvTopology, +} + +impl KvLocalApp { + #[must_use] + pub fn nodes(nodes: usize) -> Self { + Self { + deployment: KvTopology::new(nodes), + } + } + + #[must_use] + pub fn deployment(&self) -> KvTopology { + self.deployment.clone() + } +} + +#[async_trait] +impl AppDeployment for KvLocalApp { + type Handle = LocalAppCluster; + + async fn deploy(self, ctx: &mut DeployContext) -> Result { + ctx.deploy_local_cluster::(self.deployment).await + } +} + impl ClusterNodeConfigApplication for KvEnv { type ConfigError = Error; diff --git a/examples/kvstore/testing/integration/src/local_env.rs b/examples/kvstore/testing/integration/src/local_env.rs index 82e8778..e643d3f 100644 --- a/examples/kvstore/testing/integration/src/local_env.rs +++ b/examples/kvstore/testing/integration/src/local_env.rs @@ -1,8 +1,9 @@ -use std::collections::HashMap; +use std::{collections::HashMap, path::PathBuf, sync::Arc}; use testing_framework_core::scenario::{DynError, StartNodeOptions}; use testing_framework_runner_local::{ - LocalBinaryApp, LocalNodePorts, LocalPeerNode, LocalProcessSpec, + BinaryProviderRef, BuildBinaryProvider, BuildCommand, EnvBinaryProvider, + FallbackBinaryProvider, LocalBinaryApp, LocalNodePorts, LocalPeerNode, LocalProcessSpec, build_local_cluster_node_config, yaml_node_config, }; @@ -28,7 +29,9 @@ impl LocalBinaryApp for KvEnv { } fn local_process_spec() -> LocalProcessSpec { - LocalProcessSpec::new("KVSTORE_NODE_BIN").with_rust_log("kvstore_node=info") + LocalProcessSpec::new("KVSTORE_NODE_BIN") + .with_binary_provider(kvstore_binary_provider()) + .with_rust_log("kvstore_node=info") } fn render_local_config(config: &KvNodeConfig) -> Result, DynError> { @@ -39,3 +42,31 @@ impl LocalBinaryApp for KvEnv { config.http_port } } + +fn kvstore_binary_provider() -> FallbackBinaryProvider { + let workspace = workspace_root(); + let providers: [BinaryProviderRef; 2] = [ + Arc::new(EnvBinaryProvider::new("KVSTORE_NODE_BIN")), + Arc::new(BuildBinaryProvider { + command: BuildCommand::new("cargo").with_args([ + "build", + "-p", + "kvstore-node", + "--bin", + "kvstore-node", + ]), + output_path: PathBuf::from(format!( + "target/debug/kvstore-node{}", + std::env::consts::EXE_SUFFIX + )), + working_dir: Some(workspace), + lock_dir: None, + }), + ]; + + FallbackBinaryProvider::new(providers) +} + +fn workspace_root() -> PathBuf { + PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../../..") +} diff --git a/examples/kvstore/testing/integration/src/scenario.rs b/examples/kvstore/testing/integration/src/scenario.rs index 5a0e770..39a7e1b 100644 --- a/examples/kvstore/testing/integration/src/scenario.rs +++ b/examples/kvstore/testing/integration/src/scenario.rs @@ -1,15 +1,22 @@ +use testing_framework_app::AppScenarioBuilderExt; use testing_framework_core::scenario::ScenarioBuilder; -use crate::{KvEnv, KvTopology}; +use crate::{KvEnv, KvExistingClusterApp, KvTopology}; pub type KvScenarioBuilder = ScenarioBuilder; pub trait KvBuilderExt: Sized { fn deployment_with(f: impl FnOnce(KvTopology) -> KvTopology) -> Self; + + fn with_existing_kvstore_app(app: KvExistingClusterApp) -> Self; } impl KvBuilderExt for KvScenarioBuilder { fn deployment_with(f: impl FnOnce(KvTopology) -> KvTopology) -> Self { KvScenarioBuilder::with_deployment(f(KvTopology::new(3))) } + + fn with_existing_kvstore_app(app: KvExistingClusterApp) -> Self { + KvScenarioBuilder::with_deployment(app.topology()).with_app(app) + } } diff --git a/examples/kvstore/testing/workloads/Cargo.toml b/examples/kvstore/testing/workloads/Cargo.toml index e400c2a..14960d6 100644 --- a/examples/kvstore/testing/workloads/Cargo.toml +++ b/examples/kvstore/testing/workloads/Cargo.toml @@ -7,6 +7,7 @@ version.workspace = true [dependencies] kvstore-node = { path = "../../kvstore-node" } kvstore-runtime-ext = { path = "../integration" } +testing-framework-app = { workspace = true } testing-framework-core = { workspace = true } async-trait = { workspace = true } diff --git a/examples/kvstore/testing/workloads/src/lib.rs b/examples/kvstore/testing/workloads/src/lib.rs index 23ac375..618afff 100644 --- a/examples/kvstore/testing/workloads/src/lib.rs +++ b/examples/kvstore/testing/workloads/src/lib.rs @@ -2,5 +2,7 @@ mod expectations; mod write; pub use expectations::KvConverges; -pub use kvstore_runtime_ext::{KvBuilderExt, KvEnv, KvScenarioBuilder, KvTopology}; -pub use write::KvWriteWorkload; +pub use kvstore_runtime_ext::{ + KvBuilderExt, KvEnv, KvExistingClusterApp, KvScenarioBuilder, KvStoreCluster, KvTopology, +}; +pub use write::{KvClusterAccessible, KvWriteWorkload}; diff --git a/examples/kvstore/testing/workloads/src/write.rs b/examples/kvstore/testing/workloads/src/write.rs index ec65a05..073a21b 100644 --- a/examples/kvstore/testing/workloads/src/write.rs +++ b/examples/kvstore/testing/workloads/src/write.rs @@ -1,8 +1,9 @@ use std::time::Duration; use async_trait::async_trait; -use kvstore_runtime_ext::KvEnv; +use kvstore_runtime_ext::{KvEnv, KvStoreCluster}; use serde::{Deserialize, Serialize}; +use testing_framework_app::AppRunContextExt; use testing_framework_core::scenario::{DynError, RunContext, Workload}; use tracing::info; @@ -68,6 +69,18 @@ impl Default for KvWriteWorkload { } } +#[derive(Clone)] +pub struct KvClusterAccessible { + expected_nodes: usize, +} + +impl KvClusterAccessible { + #[must_use] + pub const fn new(expected_nodes: usize) -> Self { + Self { expected_nodes } + } +} + #[async_trait] impl Workload for KvWriteWorkload { fn name(&self) -> &str { @@ -126,6 +139,42 @@ impl Workload for KvWriteWorkload { } } +#[async_trait] +impl Workload for KvClusterAccessible { + fn name(&self) -> &str { + "kv_cluster_accessible" + } + + async fn start(&self, ctx: &RunContext) -> Result<(), DynError> { + let cluster = ctx.require_app::()?; + let client_count = cluster.clients().len(); + + if cluster.node_count() != self.expected_nodes { + return Err(format!( + "kv app topology has {} nodes, expected {}", + cluster.node_count(), + self.expected_nodes + ) + .into()); + } + + if client_count != self.expected_nodes { + return Err(format!( + "kv app handle has {client_count} clients, expected {}", + self.expected_nodes + ) + .into()); + } + + info!( + nodes = self.expected_nodes, + "kv app handle is accessible from workload" + ); + + Ok(()) + } +} + fn compute_interval(rate_per_sec: usize) -> Option { if rate_per_sec == 0 { return None; diff --git a/examples/multi_app/README.md b/examples/multi_app/README.md new file mode 100644 index 0000000..abec9bb --- /dev/null +++ b/examples/multi_app/README.md @@ -0,0 +1,56 @@ +# Multi-App Examples + +This directory shows the canonical app-layer pattern for composed systems. + +Use this shape when a scenario needs several apps or resources that should feel +like one system to the workload: + +```rust +let mut scenario = AppHost::scenario() + .with_app(ExampleStackApp::new()) + .with_workload(ExampleStackWorkload::new()) + .build()?; +``` + +The stack app deploys child apps, exposes their typed handles, and returns a +composed stack handle: + +```rust +let store = StoreHandle::new(ctx.deploy_and_expose(self.store).await?); +let consensus = ConsensusHandle::new(ctx.deploy_and_expose(self.consensus).await?); +let wallet = WalletHandle::new(store.clone(), consensus.clone()); +let stack = ExampleStackHandle::new(store.clone(), consensus.clone(), wallet.clone()); + +ctx.expose(store)?; +ctx.expose(consensus)?; +ctx.expose(wallet)?; +ctx.expose(stack.clone())?; +``` + +Workloads then request concrete handles: + +```rust +let stack = ctx.require_app::()?; +let wallet = ctx.require_app::()?; +``` + +Resource lifecycle comes from the TF adapter used by a child deployment: + +- `LocalProcessApp` manages one local binary process. +- `LocalAppCluster` manages a uniform local cluster. + +Attached and external sources are not registered again at the app layer. The +outer scenario resolves them through its existing source providers, and the app +deployment receives the resulting node clients and control profile through +`DeployContext`. When the active deployer provides node control or cluster +readiness, the same handles are available through `DeployContext::node_control` +and `DeployContext::cluster_wait`; the app layer does not create replacements. + +Managed adapters register cleanup during deployment. Cleanup runs in reverse +acquisition order and does not depend on the last handle clone being dropped. +An app-specific deployment only describes composition; it does not implement a +second lifecycle interface. + +For a single uniform cluster, the core `ScenarioBuilder` flow remains +valid. For composed systems, prefer this app-layer shape instead of building a +fake outer cluster or adding app-specific code to TF. diff --git a/examples/multi_app/examples/Cargo.toml b/examples/multi_app/examples/Cargo.toml new file mode 100644 index 0000000..bd40515 --- /dev/null +++ b/examples/multi_app/examples/Cargo.toml @@ -0,0 +1,22 @@ +[package] +edition.workspace = true +license.workspace = true +name = "multi-app-examples" +version.workspace = true + +[[bin]] +name = "multi_app_typed_stack_smoke" +path = "src/bin/typed_stack_smoke.rs" + +[dependencies] +anyhow = "1.0" +async-trait = { workspace = true } +kvstore-runtime-ext = { path = "../../kvstore/testing/integration" } +openraft-kv-runtime-ext = { path = "../../openraft_kv/testing/integration" } +openraft-kv-runtime-workloads = { path = "../../openraft_kv/testing/workloads" } +serde = { workspace = true } +testing-framework-app = { workspace = true } +testing-framework-core = { workspace = true } +tokio = { workspace = true, features = ["full"] } +tracing = { workspace = true } +tracing-subscriber = { version = "0.3", features = ["env-filter"] } diff --git a/examples/multi_app/examples/src/bin/typed_stack_smoke.rs b/examples/multi_app/examples/src/bin/typed_stack_smoke.rs new file mode 100644 index 0000000..a857b0b --- /dev/null +++ b/examples/multi_app/examples/src/bin/typed_stack_smoke.rs @@ -0,0 +1,269 @@ +use std::time::Duration; + +use async_trait::async_trait; +use kvstore_runtime_ext::{KvEnv, KvLocalApp}; +use openraft_kv_runtime_ext::{OpenRaftKvEnv, OpenRaftKvLocalApp}; +use openraft_kv_runtime_workloads::{ + OpenRaftMembership, ensure_cluster_size, expected_kv, resolve_client_for_node, wait_for_leader, + wait_for_membership, wait_for_replication, write_batch, +}; +use serde::{Deserialize, Serialize}; +use testing_framework_app::{ + AppDeployment, AppHost, AppHostEnv, AppHostLocalDeployer, AppRunContextExt, + AppScenarioBuilderExt, DeployContext, LocalAppCluster, +}; +use testing_framework_core::scenario::{Deployer, DynError, RunContext, Workload}; +use tracing::info; + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + tracing_subscriber::fmt() + .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) + .init(); + + let mut scenario = AppHost::scenario() + .with_app(ExampleStackApp::new()) + .with_run_duration(Duration::from_secs(5)) + .with_workload(ExampleStackWorkload::new()) + .build()?; + + let deployer = AppHostLocalDeployer::default(); + let runner = deployer.deploy(&scenario).await?; + runner.run(&mut scenario).await?; + + Ok(()) +} + +#[derive(Clone)] +struct ExampleStackApp { + store: KvLocalApp, + consensus: OpenRaftKvLocalApp, +} + +impl ExampleStackApp { + fn new() -> Self { + Self { + store: KvLocalApp::nodes(2), + consensus: OpenRaftKvLocalApp::nodes(3), + } + } +} + +#[async_trait] +impl AppDeployment for ExampleStackApp { + type Handle = ExampleStackHandle; + + async fn deploy(self, ctx: &mut DeployContext) -> Result { + let store = StoreHandle::new(ctx.deploy_and_expose(self.store).await?); + let consensus = ConsensusHandle::new(ctx.deploy_and_expose(self.consensus).await?); + let wallet = WalletHandle::new(store.clone(), consensus.clone()); + let stack = ExampleStackHandle::new(store.clone(), consensus.clone(), wallet.clone()); + + ctx.expose(store)?; + ctx.expose(consensus)?; + ctx.expose(wallet)?; + ctx.expose(stack.clone())?; + + Ok(stack) + } +} + +#[derive(Clone)] +struct ExampleStackHandle { + store: StoreHandle, + consensus: ConsensusHandle, + wallet: WalletHandle, +} + +impl ExampleStackHandle { + const fn new(store: StoreHandle, consensus: ConsensusHandle, wallet: WalletHandle) -> Self { + Self { + store, + consensus, + wallet, + } + } + + const fn store(&self) -> &StoreHandle { + &self.store + } + + const fn consensus(&self) -> &ConsensusHandle { + &self.consensus + } + + const fn wallet(&self) -> &WalletHandle { + &self.wallet + } +} + +#[derive(Clone)] +struct StoreHandle { + cluster: LocalAppCluster, +} + +impl StoreHandle { + const fn new(cluster: LocalAppCluster) -> Self { + Self { cluster } + } + + fn node_count(&self) -> usize { + self.cluster.node_count() + } + + async fn put(&self, key: &str, value: &str) -> Result<(), DynError> { + let Some(client) = self.cluster.first_client() else { + return Err("store handle has no kv clients".into()); + }; + + let response: KvPutResponse = client + .put( + key, + &KvPutRequest { + value: value.to_owned(), + expected_version: None, + }, + ) + .await?; + + if !response.applied { + return Err(format!("store write for {key} was rejected").into()); + } + + Ok(()) + } +} + +#[derive(Clone)] +struct ConsensusHandle { + cluster: LocalAppCluster, +} + +impl ConsensusHandle { + const fn new(cluster: LocalAppCluster) -> Self { + Self { cluster } + } + + fn node_count(&self) -> usize { + self.cluster.node_count() + } + + async fn bootstrap_and_write(&self, prefix: &str, writes: usize) -> Result<(), DynError> { + let clients = self.cluster.clients(); + + ensure_cluster_size(&clients, self.node_count())?; + clients[0].init_self().await?; + + let leader_id = wait_for_leader(&clients, Duration::from_secs(30), None).await?; + let membership = OpenRaftMembership::discover(&clients).await?; + let leader = resolve_client_for_node(&clients, leader_id, Duration::from_secs(30)).await?; + + for learner in membership.learner_targets(leader_id) { + leader + .add_learner(learner.node_id, &learner.public_addr) + .await?; + } + + let voter_ids = membership.voter_ids(); + leader.change_membership(voter_ids.iter().copied()).await?; + wait_for_membership(&clients, &voter_ids, Duration::from_secs(30)).await?; + + write_batch(&leader, prefix, 0, writes).await?; + wait_for_replication( + &clients, + &expected_kv(prefix, writes), + Duration::from_secs(30), + ) + .await?; + + Ok(()) + } +} + +#[derive(Clone)] +struct WalletHandle { + store: StoreHandle, + consensus: ConsensusHandle, +} + +impl WalletHandle { + const fn new(store: StoreHandle, consensus: ConsensusHandle) -> Self { + Self { store, consensus } + } + + async fn submit_transfer(&self, id: &str) -> Result<(), DynError> { + self.store.put("/kv/wallet-transfer", id).await?; + self.consensus + .bootstrap_and_write("wallet-transfer", 3) + .await?; + + Ok(()) + } +} + +#[derive(Clone)] +struct ExampleStackWorkload; + +impl ExampleStackWorkload { + const fn new() -> Self { + Self + } +} + +#[async_trait] +impl Workload for ExampleStackWorkload { + fn name(&self) -> &str { + "multi_app_typed_stack_smoke" + } + + async fn start(&self, ctx: &RunContext) -> Result<(), DynError> { + let stack = ctx.require_app::()?; + let store = ctx.require_app::()?; + let consensus = ctx.require_app::()?; + let wallet = ctx.require_app::()?; + + ensure_stack_handles_match(&stack, &store, &consensus, &wallet)?; + store.put("/kv/typed-stack-smoke", "store-ready").await?; + wallet.submit_transfer("transfer-1").await?; + + info!( + store_nodes = store.node_count(), + consensus_nodes = consensus.node_count(), + "typed multi-app stack handles are available to workloads" + ); + + Ok(()) + } +} + +fn ensure_stack_handles_match( + stack: &ExampleStackHandle, + store: &StoreHandle, + consensus: &ConsensusHandle, + wallet: &WalletHandle, +) -> Result<(), DynError> { + if stack.store().node_count() != store.node_count() { + return Err("stack store handle does not match exposed store handle".into()); + } + + if stack.consensus().node_count() != consensus.node_count() { + return Err("stack consensus handle does not match exposed consensus handle".into()); + } + + if stack.wallet().store.node_count() != wallet.store.node_count() { + return Err("stack wallet handle does not match exposed wallet handle".into()); + } + + Ok(()) +} + +#[derive(Serialize)] +struct KvPutRequest { + value: String, + expected_version: Option, +} + +#[derive(Deserialize)] +struct KvPutResponse { + applied: bool, +} diff --git a/examples/openraft_kv/examples/Cargo.toml b/examples/openraft_kv/examples/Cargo.toml index b514dc6..5b9e46b 100644 --- a/examples/openraft_kv/examples/Cargo.toml +++ b/examples/openraft_kv/examples/Cargo.toml @@ -4,6 +4,10 @@ license.workspace = true name = "openraft-kv-examples" version.workspace = true +[[bin]] +name = "openraft_kv_app_host_smoke" +path = "src/bin/app_host_smoke.rs" + [[bin]] name = "openraft_kv_basic_failover" path = "src/bin/basic_failover.rs" @@ -18,9 +22,11 @@ path = "src/bin/k8s_failover.rs" [dependencies] anyhow = "1.0" +async-trait = { workspace = true } openraft-kv-node = { path = "../openraft-kv-node" } openraft-kv-runtime-ext = { path = "../testing/integration" } openraft-kv-runtime-workloads = { path = "../testing/workloads" } +testing-framework-app = { workspace = true } testing-framework-core = { workspace = true } testing-framework-runner-k8s = { workspace = true } tokio = { workspace = true, features = ["full"] } diff --git a/examples/openraft_kv/examples/src/bin/app_host_smoke.rs b/examples/openraft_kv/examples/src/bin/app_host_smoke.rs new file mode 100644 index 0000000..94d97e6 --- /dev/null +++ b/examples/openraft_kv/examples/src/bin/app_host_smoke.rs @@ -0,0 +1,118 @@ +use std::time::Duration; + +use async_trait::async_trait; +use openraft_kv_runtime_ext::{OpenRaftKvEnv, OpenRaftKvLocalApp}; +use openraft_kv_runtime_workloads::{ + OpenRaftMembership, ensure_cluster_size, expected_kv, resolve_client_for_node, wait_for_leader, + wait_for_membership, wait_for_replication, write_batch, +}; +use testing_framework_app::{ + AppHost, AppHostEnv, AppHostLocalDeployer, AppRunContextExt, AppScenarioBuilderExt, + LocalAppCluster, +}; +use testing_framework_core::scenario::{Deployer, DynError, RunContext, Workload}; +use tracing::info; + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + tracing_subscriber::fmt() + .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) + .init(); + + let mut scenario = AppHost::scenario() + .with_app(OpenRaftKvLocalApp::nodes(3)) + .with_run_duration(Duration::from_secs(5)) + .with_workload(OpenRaftKvAppHostSmoke::new(3)) + .build()?; + + let deployer = AppHostLocalDeployer::default(); + let runner = deployer.deploy(&scenario).await?; + runner.run(&mut scenario).await?; + + Ok(()) +} + +#[derive(Clone)] +struct OpenRaftKvAppHostSmoke { + expected_nodes: usize, + timeout: Duration, +} + +impl OpenRaftKvAppHostSmoke { + const fn new(expected_nodes: usize) -> Self { + Self { + expected_nodes, + timeout: Duration::from_secs(30), + } + } +} + +#[async_trait] +impl Workload for OpenRaftKvAppHostSmoke { + fn name(&self) -> &str { + "openraft_kv_app_host_smoke" + } + + async fn start(&self, ctx: &RunContext) -> Result<(), DynError> { + let cluster = ctx.require_app::>()?; + + ensure_cluster_shape(&cluster, self.expected_nodes)?; + bootstrap_voter_cluster(&cluster, self.timeout).await?; + + info!( + nodes = self.expected_nodes, + "openraft app host cluster exposes app-local clients and process access" + ); + + Ok(()) + } +} + +fn ensure_cluster_shape( + cluster: &LocalAppCluster, + expected_nodes: usize, +) -> Result<(), DynError> { + if cluster.node_count() != expected_nodes { + return Err(format!("openraft app host expected {expected_nodes} nodes").into()); + } + + if cluster.node_client("node-0").is_none() { + return Err("openraft app host cannot access node-0 client".into()); + } + + if cluster.node_pid("node-0").is_none() { + return Err("openraft app host cannot access node-0 process id".into()); + } + + ensure_cluster_size(&cluster.clients(), expected_nodes)?; + + Ok(()) +} + +async fn bootstrap_voter_cluster( + cluster: &LocalAppCluster, + timeout: Duration, +) -> Result<(), DynError> { + let clients = cluster.clients(); + + clients[0].init_self().await?; + + let initial_leader = wait_for_leader(&clients, timeout, None).await?; + let membership = OpenRaftMembership::discover(&clients).await?; + let leader = resolve_client_for_node(&clients, initial_leader, timeout).await?; + + for learner in membership.learner_targets(initial_leader) { + leader + .add_learner(learner.node_id, &learner.public_addr) + .await?; + } + + let voter_ids = membership.voter_ids(); + leader.change_membership(voter_ids.iter().copied()).await?; + wait_for_membership(&clients, &voter_ids, timeout).await?; + + write_batch(&leader, "app-host-raft-key", 0, 3).await?; + wait_for_replication(&clients, &expected_kv("app-host-raft-key", 3), timeout).await?; + + Ok(()) +} diff --git a/examples/openraft_kv/examples/src/lib.rs b/examples/openraft_kv/examples/src/lib.rs index 8e19a67..1101c38 100644 --- a/examples/openraft_kv/examples/src/lib.rs +++ b/examples/openraft_kv/examples/src/lib.rs @@ -1,7 +1,11 @@ use std::time::Duration; -use openraft_kv_runtime_ext::{OpenRaftKvBuilderExt, OpenRaftKvEnv, OpenRaftKvScenarioBuilder}; -use openraft_kv_runtime_workloads::{OpenRaftKvConverges, OpenRaftKvFailoverWorkload}; +use openraft_kv_runtime_ext::{ + OpenRaftKvBuilderExt, OpenRaftKvEnv, OpenRaftKvExistingClusterApp, OpenRaftKvScenarioBuilder, +}; +use openraft_kv_runtime_workloads::{ + OpenRaftKvClusterAccessible, OpenRaftKvConverges, OpenRaftKvFailoverWorkload, +}; use testing_framework_core::scenario::{NodeControlCapability, Scenario}; /// Number of writes issued before the leader restart. @@ -19,23 +23,23 @@ pub fn build_failover_scenario( run_duration: Duration, workload_timeout: Duration, ) -> anyhow::Result> { - Ok( - OpenRaftKvScenarioBuilder::deployment_with(|deployment| deployment) - .with_cluster_observer() - .enable_node_control() - .with_run_duration(run_duration) - .with_workload( - OpenRaftKvFailoverWorkload::new() - .first_batch(INITIAL_WRITE_BATCH) - .second_batch(SECOND_WRITE_BATCH) - .timeout(workload_timeout) - .key_prefix(RAFT_KEY_PREFIX), - ) - .with_expectation( - OpenRaftKvConverges::new(TOTAL_WRITES) - .timeout(run_duration) - .key_prefix(RAFT_KEY_PREFIX), - ) - .build()?, + Ok(OpenRaftKvScenarioBuilder::with_existing_openraft_kv_app( + OpenRaftKvExistingClusterApp::nodes(3), ) + .enable_node_control() + .with_run_duration(run_duration) + .with_workload(OpenRaftKvClusterAccessible::new(3)) + .with_workload( + OpenRaftKvFailoverWorkload::new() + .first_batch(INITIAL_WRITE_BATCH) + .second_batch(SECOND_WRITE_BATCH) + .timeout(workload_timeout) + .key_prefix(RAFT_KEY_PREFIX), + ) + .with_expectation( + OpenRaftKvConverges::new(TOTAL_WRITES) + .timeout(run_duration) + .key_prefix(RAFT_KEY_PREFIX), + ) + .build()?) } diff --git a/examples/openraft_kv/testing/integration/Cargo.toml b/examples/openraft_kv/testing/integration/Cargo.toml index ef40e34..e058731 100644 --- a/examples/openraft_kv/testing/integration/Cargo.toml +++ b/examples/openraft_kv/testing/integration/Cargo.toml @@ -8,6 +8,7 @@ version.workspace = true async-trait = { workspace = true } openraft-kv-node = { path = "../../openraft-kv-node" } reqwest = { workspace = true } +testing-framework-app = { workspace = true } testing-framework-core = { workspace = true } testing-framework-runner-compose = { workspace = true } testing-framework-runner-k8s = { workspace = true } diff --git a/examples/openraft_kv/testing/integration/src/app.rs b/examples/openraft_kv/testing/integration/src/app.rs index bc2c7f1..6b37059 100644 --- a/examples/openraft_kv/testing/integration/src/app.rs +++ b/examples/openraft_kv/testing/integration/src/app.rs @@ -1,9 +1,14 @@ use std::io::Error; +use async_trait::async_trait; use openraft_kv_node::{OpenRaftKvClient, OpenRaftKvNodeConfig}; -use testing_framework_core::scenario::{ - Application, ClusterNodeConfigApplication, ClusterNodeView, ClusterPeerView, DynError, - NodeAccess, serialize_cluster_yaml_config, +use testing_framework_app::{AppDeployment, AppHostEnv, DeployContext, LocalAppCluster}; +use testing_framework_core::{ + scenario::{ + Application, ClusterNodeConfigApplication, ClusterNodeView, ClusterPeerView, DynError, + NodeAccess, NodeClients, serialize_cluster_yaml_config, + }, + topology::DeploymentDescriptor, }; /// Three-node topology used by the OpenRaft example scenarios. @@ -57,3 +62,116 @@ impl ClusterNodeConfigApplication for OpenRaftKvEnv { serialize_cluster_yaml_config(config).map_err(Error::other) } } + +/// Runtime handle exposed by the OpenRaft app deployment. +#[derive(Clone)] +pub struct OpenRaftKvCluster { + deployment: OpenRaftKvTopology, + node_clients: NodeClients, +} + +impl OpenRaftKvCluster { + #[must_use] + pub const fn new( + deployment: OpenRaftKvTopology, + node_clients: NodeClients, + ) -> Self { + Self { + deployment, + node_clients, + } + } + + #[must_use] + pub fn topology(&self) -> &OpenRaftKvTopology { + &self.deployment + } + + #[must_use] + pub fn node_count(&self) -> usize { + self.deployment.node_count() + } + + #[must_use] + pub fn clients(&self) -> Vec { + self.node_clients.snapshot() + } + + pub async fn states(&self) -> Result, DynError> { + let clients = self.clients(); + let mut states = Vec::with_capacity(clients.len()); + + for client in clients { + states.push(client.state().await?); + } + + states.sort_by_key(|state| state.node_id); + + Ok(states) + } +} + +/// App preset for the OpenRaft key-value example. +#[derive(Clone)] +pub struct OpenRaftKvExistingClusterApp { + topology: OpenRaftKvTopology, +} + +impl OpenRaftKvExistingClusterApp { + #[must_use] + pub fn nodes(nodes: usize) -> Self { + Self { + topology: OpenRaftKvTopology::new(nodes), + } + } + + #[must_use] + pub fn topology(&self) -> OpenRaftKvTopology { + self.topology.clone() + } +} + +#[async_trait] +impl AppDeployment for OpenRaftKvExistingClusterApp { + type Handle = OpenRaftKvCluster; + + async fn deploy( + self, + ctx: &mut DeployContext, + ) -> Result { + Ok(OpenRaftKvCluster::new( + ctx.deployment().clone(), + ctx.node_clients().clone(), + )) + } +} + +/// Local app preset that starts its own OpenRaft cluster. +#[derive(Clone)] +pub struct OpenRaftKvLocalApp { + deployment: OpenRaftKvTopology, +} + +impl OpenRaftKvLocalApp { + #[must_use] + pub fn nodes(nodes: usize) -> Self { + Self { + deployment: OpenRaftKvTopology::new(nodes), + } + } + + #[must_use] + pub fn deployment(&self) -> OpenRaftKvTopology { + self.deployment.clone() + } +} + +#[async_trait] +impl AppDeployment for OpenRaftKvLocalApp { + type Handle = LocalAppCluster; + + async fn deploy(self, ctx: &mut DeployContext) -> Result { + ctx.deploy_local_cluster::(self.deployment) + .await + } +} diff --git a/examples/openraft_kv/testing/integration/src/local_env.rs b/examples/openraft_kv/testing/integration/src/local_env.rs index c01183f..e65f440 100644 --- a/examples/openraft_kv/testing/integration/src/local_env.rs +++ b/examples/openraft_kv/testing/integration/src/local_env.rs @@ -1,4 +1,8 @@ -use std::collections::{BTreeMap, HashMap}; +use std::{ + collections::{BTreeMap, HashMap}, + path::PathBuf, + sync::Arc, +}; use openraft_kv_node::OpenRaftKvNodeConfig; use testing_framework_core::{ @@ -6,7 +10,8 @@ use testing_framework_core::{ topology::DeploymentDescriptor, }; use testing_framework_runner_local::{ - BuiltNodeConfig, LocalDeployerEnv, LocalNodePorts, LocalProcessSpec, NodeConfigEntry, + BinaryProviderRef, BuildBinaryProvider, BuildCommand, BuiltNodeConfig, EnvBinaryProvider, + FallbackBinaryProvider, LocalDeployerEnv, LocalNodePorts, LocalProcessSpec, NodeConfigEntry, reserve_local_node_ports, yaml_node_config, }; @@ -80,7 +85,11 @@ impl LocalDeployerEnv for OpenRaftKvEnv { } fn local_process_spec() -> Option { - Some(LocalProcessSpec::new("OPENRAFT_KV_NODE_BIN").with_rust_log("info")) + Some( + LocalProcessSpec::new("OPENRAFT_KV_NODE_BIN") + .with_binary_provider(openraft_binary_provider()) + .with_rust_log("info"), + ) } fn render_local_config(config: &OpenRaftKvNodeConfig) -> Result, DynError> { @@ -92,6 +101,34 @@ impl LocalDeployerEnv for OpenRaftKvEnv { } } +fn openraft_binary_provider() -> FallbackBinaryProvider { + let workspace = workspace_root(); + let providers: [BinaryProviderRef; 2] = [ + Arc::new(EnvBinaryProvider::new("OPENRAFT_KV_NODE_BIN")), + Arc::new(BuildBinaryProvider { + command: BuildCommand::new("cargo").with_args([ + "build", + "-p", + "openraft-kv-node", + "--bin", + "openraft-kv-node", + ]), + output_path: PathBuf::from(format!( + "target/debug/openraft-kv-node{}", + std::env::consts::EXE_SUFFIX + )), + working_dir: Some(workspace), + lock_dir: None, + }), + ]; + + FallbackBinaryProvider::new(providers) +} + +fn workspace_root() -> PathBuf { + PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../../..") +} + fn local_node_config( index: usize, network_port: u16, diff --git a/examples/openraft_kv/testing/integration/src/scenario.rs b/examples/openraft_kv/testing/integration/src/scenario.rs index d2de604..432d6e6 100644 --- a/examples/openraft_kv/testing/integration/src/scenario.rs +++ b/examples/openraft_kv/testing/integration/src/scenario.rs @@ -1,7 +1,9 @@ +use testing_framework_app::AppScenarioBuilderExt; use testing_framework_core::scenario::{CoreBuilderExt, ScenarioBuilder}; use crate::{ - OpenRaftClusterObserver, OpenRaftKvEnv, OpenRaftKvTopology, openraft_cluster_source_provider, + OpenRaftClusterObserver, OpenRaftKvEnv, OpenRaftKvExistingClusterApp, OpenRaftKvTopology, + openraft_cluster_source_provider, }; /// Scenario builder alias used by the OpenRaft example binaries. @@ -15,6 +17,10 @@ pub trait OpenRaftKvBuilderExt: Sized { /// Attaches the default OpenRaft cluster observer to the scenario. fn with_cluster_observer(self) -> Self; + + /// Starts from an OpenRaft app preset and exposes its cluster handle to + /// workloads. + fn with_existing_openraft_kv_app(app: OpenRaftKvExistingClusterApp) -> Self; } impl OpenRaftKvBuilderExt for OpenRaftKvScenarioBuilder { @@ -29,4 +35,10 @@ impl OpenRaftKvBuilderExt for OpenRaftKvScenarioBuilder { OpenRaftClusterObserver::config(), ) } + + fn with_existing_openraft_kv_app(app: OpenRaftKvExistingClusterApp) -> Self { + OpenRaftKvScenarioBuilder::with_deployment(app.topology()) + .with_app(app) + .with_cluster_observer() + } } diff --git a/examples/openraft_kv/testing/workloads/Cargo.toml b/examples/openraft_kv/testing/workloads/Cargo.toml index 0682b55..5e03aaf 100644 --- a/examples/openraft_kv/testing/workloads/Cargo.toml +++ b/examples/openraft_kv/testing/workloads/Cargo.toml @@ -9,6 +9,7 @@ anyhow = "1.0" async-trait = { workspace = true } openraft-kv-node = { path = "../../openraft-kv-node" } openraft-kv-runtime-ext = { path = "../integration" } +testing-framework-app = { workspace = true } testing-framework-core = { workspace = true } thiserror = "2.0" tokio = { workspace = true, features = ["full"] } diff --git a/examples/openraft_kv/testing/workloads/src/handle_access.rs b/examples/openraft_kv/testing/workloads/src/handle_access.rs new file mode 100644 index 0000000..758959c --- /dev/null +++ b/examples/openraft_kv/testing/workloads/src/handle_access.rs @@ -0,0 +1,63 @@ +use async_trait::async_trait; +use openraft_kv_runtime_ext::{OpenRaftKvCluster, OpenRaftKvEnv}; +use testing_framework_app::AppRunContextExt; +use testing_framework_core::scenario::{DynError, RunContext, Workload}; +use tracing::info; + +#[derive(Clone)] +pub struct OpenRaftKvClusterAccessible { + expected_nodes: usize, +} + +impl OpenRaftKvClusterAccessible { + #[must_use] + pub const fn new(expected_nodes: usize) -> Self { + Self { expected_nodes } + } +} + +#[async_trait] +impl Workload for OpenRaftKvClusterAccessible { + fn name(&self) -> &str { + "openraft_kv_cluster_accessible" + } + + async fn start(&self, ctx: &RunContext) -> Result<(), DynError> { + let cluster = ctx.require_app::()?; + let states = cluster.states().await?; + let client_count = cluster.clients().len(); + + if cluster.node_count() != self.expected_nodes { + return Err(format!( + "openraft app topology has {} nodes, expected {}", + cluster.node_count(), + self.expected_nodes + ) + .into()); + } + + if client_count != self.expected_nodes { + return Err(format!( + "openraft app handle has {client_count} clients, expected {}", + self.expected_nodes + ) + .into()); + } + + if states.len() != self.expected_nodes { + return Err(format!( + "openraft app handle read {} node states, expected {}", + states.len(), + self.expected_nodes + ) + .into()); + } + + info!( + nodes = self.expected_nodes, + "openraft app handle is accessible from workload" + ); + + Ok(()) + } +} diff --git a/examples/openraft_kv/testing/workloads/src/lib.rs b/examples/openraft_kv/testing/workloads/src/lib.rs index 02bd5a8..4629c5d 100644 --- a/examples/openraft_kv/testing/workloads/src/lib.rs +++ b/examples/openraft_kv/testing/workloads/src/lib.rs @@ -1,11 +1,14 @@ mod convergence; mod failover; +mod handle_access; mod support; /// Replication expectation used by the OpenRaft example binaries. pub use convergence::OpenRaftKvConverges; /// Failover workload used by the OpenRaft example binaries. pub use failover::OpenRaftKvFailoverWorkload; +/// Workload that proves app handles are available to OpenRaft workloads. +pub use handle_access::OpenRaftKvClusterAccessible; /// Shared cluster helpers used by the OpenRaft workload and manual k8s example. pub use support::{ FULL_VOTER_SET, OpenRaftClusterError, OpenRaftMembership, ensure_cluster_size, expected_kv, diff --git a/testing-framework/app/Cargo.toml b/testing-framework/app/Cargo.toml new file mode 100644 index 0000000..44ce590 --- /dev/null +++ b/testing-framework/app/Cargo.toml @@ -0,0 +1,21 @@ +[package] +description.workspace = true +edition.workspace = true +license.workspace = true +name = "testing-framework-app" +readme.workspace = true +repository.workspace = true +version.workspace = true + +[lints] +workspace = true + +[dependencies] +async-trait = { workspace = true } +testing-framework-core = { workspace = true } +testing-framework-runner-local = { workspace = true } +thiserror = { workspace = true } +tokio = { features = ["rt", "sync"], workspace = true } + +[dev-dependencies] +tokio = { features = ["macros", "rt"], workspace = true } diff --git a/testing-framework/app/src/cleanup.rs b/testing-framework/app/src/cleanup.rs new file mode 100644 index 0000000..039f1b8 --- /dev/null +++ b/testing-framework/app/src/cleanup.rs @@ -0,0 +1,81 @@ +use testing_framework_core::scenario::internal::CleanupGuard; + +/// LIFO cleanup owned by one application deployment preparation. +/// +/// Managed adapters register guards here as soon as they acquire a resource. +/// If deployment fails, dropping the stack releases everything already +/// acquired. On success, the stack is transferred to the scenario runner. +#[derive(Default)] +pub(crate) struct AppCleanupStack { + guards: Vec>, +} + +impl AppCleanupStack { + pub(crate) fn push(&mut self, guard: Box) { + self.guards.push(guard); + } + + pub(crate) fn into_guard(self) -> Option> { + if self.guards.is_empty() { + None + } else { + Some(Box::new(self)) + } + } + + fn cleanup_all(&mut self) { + while let Some(guard) = self.guards.pop() { + guard.cleanup(); + } + } +} + +impl CleanupGuard for AppCleanupStack { + fn cleanup(mut self: Box) { + self.cleanup_all(); + } +} + +impl Drop for AppCleanupStack { + fn drop(&mut self) { + self.cleanup_all(); + } +} + +#[cfg(test)] +mod tests { + use std::sync::{Arc, Mutex}; + + use testing_framework_core::scenario::internal::CleanupGuard; + + use super::AppCleanupStack; + + struct Probe { + label: &'static str, + order: Arc>>, + } + + impl CleanupGuard for Probe { + fn cleanup(self: Box) { + self.order.lock().unwrap().push(self.label); + } + } + + #[test] + fn managed_resources_clean_up_in_reverse_acquisition_order() { + let order = Arc::new(Mutex::new(Vec::new())); + let mut cleanup = AppCleanupStack::default(); + cleanup.push(Box::new(Probe { + label: "dependency", + order: Arc::clone(&order), + })); + cleanup.push(Box::new(Probe { + label: "dependent", + order: Arc::clone(&order), + })); + + drop(cleanup); + + assert_eq!(*order.lock().unwrap(), ["dependent", "dependency"]); + } +} diff --git a/testing-framework/app/src/context.rs b/testing-framework/app/src/context.rs new file mode 100644 index 0000000..81b7393 --- /dev/null +++ b/testing-framework/app/src/context.rs @@ -0,0 +1,323 @@ +use testing_framework_core::scenario::{ + Application, DynError, NodeClients, internal::CleanupGuard, +}; +use testing_framework_runner_local::LocalDeployerEnv; + +use crate::{ + AppDeployError, AppDeployment, AppHandle, HandleRegistry, LocalAppCluster, + cleanup::AppCleanupStack, +}; + +/// Mutable deployment context used to compose applications and expose handles. +/// +/// A context belongs to one scenario preparation. It provides access to the +/// outer scenario deployment and node clients while collecting application +/// handles for later use by workloads. Exposed handles are dropped in reverse +/// exposure order when the scenario runtime is released. +pub struct DeployContext { + deployment: E::Deployment, + node_clients: NodeClients, + handles: HandleRegistry, + cleanup: AppCleanupStack, +} + +impl DeployContext +where + E: Application, +{ + /// Creates a context for an outer scenario deployment and its node clients. + pub fn new(deployment: E::Deployment, node_clients: NodeClients) -> Self { + Self { + deployment, + node_clients, + handles: HandleRegistry::new(), + cleanup: AppCleanupStack::default(), + } + } + + /// Deploys a child application without automatically exposing its handle. + /// + /// Use this when the returned handle is only needed while constructing a + /// higher-level application handle. + pub async fn deploy(&mut self, app: A) -> Result + where + A: AppDeployment, + { + app.deploy(self).await + } + + /// Deploys a child application and exposes a clone of its handle. + /// + /// The exposed clone makes the handle available to workloads through + /// [`crate::AppRunContextExt`]. Managed resource lifetime remains owned by + /// scenario cleanup. + pub async fn deploy_and_expose(&mut self, app: A) -> Result + where + A: AppDeployment, + { + let handle = self.deploy(app).await?; + + self.expose(handle.clone())?; + + Ok(handle) + } + + /// Exposes the default unnamed handle for its concrete type. + /// + /// Returns [`AppDeployError::DuplicateHandle`] when that type already has a + /// default handle. + pub fn expose(&mut self, handle: T) -> Result<(), AppDeployError> + where + T: AppHandle, + { + self.handles.expose(handle) + } + + /// Exposes a named handle, allowing multiple instances of the same type. + /// + /// A type and name pair must be unique within the deployment context. + pub fn expose_named( + &mut self, + name: impl Into, + handle: T, + ) -> Result<(), AppDeployError> + where + T: AppHandle, + { + self.handles.expose_named(name, handle) + } + + /// Returns a clone of the default handle for `T`, if it is exposed. + pub fn get(&self) -> Option + where + T: AppHandle, + { + self.handles.get() + } + + /// Returns a clone of the named handle for `T`, if it is exposed. + pub fn get_named(&self, name: &str) -> Option + where + T: AppHandle, + { + self.handles.get_named(name) + } + + /// Returns the default handle for `T` or a typed missing-handle error. + pub fn require(&self) -> Result + where + T: AppHandle, + { + self.handles.require() + } + + /// Returns the named handle for `T` or a typed missing-handle error. + pub fn require_named(&self, name: &str) -> Result + where + T: AppHandle, + { + self.handles.require_named(name) + } + + /// Returns whether the default handle for `T` is exposed. + pub fn contains(&self) -> bool + where + T: AppHandle, + { + self.handles.contains::() + } + + /// Borrows the handles exposed so far. + pub fn handles(&self) -> &HandleRegistry { + &self.handles + } + + /// Starts every node of an additional uniform local cluster. + /// + /// The cluster is registered with scenario cleanup before its handle is + /// returned. Cleanup stops all nodes independently of handle clones. + pub async fn deploy_local_cluster( + &mut self, + deployment: App::Deployment, + ) -> Result, DynError> + where + App: LocalDeployerEnv, + { + let cluster = LocalAppCluster::start(deployment).await?; + self.register_cleanup(cluster.cleanup_guard()); + Ok(cluster) + } + + /// Returns the outer scenario deployment descriptor. + pub fn deployment(&self) -> &E::Deployment { + &self.deployment + } + + /// Returns the outer scenario clients resolved from all configured sources. + pub fn node_clients(&self) -> &NodeClients { + &self.node_clients + } + + pub(crate) fn register_cleanup(&mut self, guard: Box) { + self.cleanup.push(guard); + } + + pub(crate) fn into_runtime_parts(self) -> (HandleRegistry, Option>) { + let Self { + handles, cleanup, .. + } = self; + (handles, cleanup.into_guard()) + } +} + +#[cfg(test)] +mod tests { + use async_trait::async_trait; + use testing_framework_core::{ + scenario::{Application, DynError, NodeClients}, + topology::DeploymentDescriptor, + }; + + use super::DeployContext; + use crate::AppDeployment; + + #[derive(Clone)] + struct TestDeployment; + + impl DeploymentDescriptor for TestDeployment { + fn node_count(&self) -> usize { + 0 + } + } + + struct TestEnv; + + #[async_trait] + impl Application for TestEnv { + type Deployment = TestDeployment; + type NodeClient = (); + type NodeConfig = (); + } + + #[derive(Clone, Debug, Eq, PartialEq)] + struct ChildHandle { + id: &'static str, + } + + struct ChildApp; + + #[async_trait] + impl AppDeployment for ChildApp { + type Handle = ChildHandle; + + async fn deploy(self, _ctx: &mut DeployContext) -> Result { + Ok(ChildHandle { id: "child" }) + } + } + + #[derive(Clone, Debug, Eq, PartialEq)] + struct SiblingHandle { + id: &'static str, + } + + struct SiblingApp; + + #[async_trait] + impl AppDeployment for SiblingApp { + type Handle = SiblingHandle; + + async fn deploy(self, _ctx: &mut DeployContext) -> Result { + Ok(SiblingHandle { id: "sibling" }) + } + } + + #[derive(Clone, Debug, Eq, PartialEq)] + struct ParentHandle { + child: ChildHandle, + } + + struct ParentApp; + + #[async_trait] + impl AppDeployment for ParentApp { + type Handle = ParentHandle; + + async fn deploy(self, ctx: &mut DeployContext) -> Result { + let child = ctx.deploy(ChildApp).await?; + let parent = ParentHandle { child }; + + ctx.expose(parent.clone())?; + + Ok(parent) + } + } + + #[tokio::test] + async fn deploy_returns_handle_without_exposing_it() { + let mut ctx = test_context(); + let child = ctx.deploy(ChildApp).await.expect("deploy child app"); + + assert_eq!(child, ChildHandle { id: "child" }); + assert!(ctx.get::().is_none()); + } + + #[tokio::test] + async fn exposed_handle_can_be_required() { + let mut ctx = test_context(); + + ctx.expose(ChildHandle { id: "child" }) + .expect("expose child"); + + let child = ctx.require::().expect("require exposed child"); + assert_eq!(child, ChildHandle { id: "child" }); + } + + #[tokio::test] + async fn nested_deployment_exposes_parent_handle_only() { + let mut ctx = test_context(); + let parent = ctx.deploy(ParentApp).await.expect("deploy parent app"); + + assert_eq!(parent.child, ChildHandle { id: "child" }); + assert!(ctx.get::().is_none()); + assert!(ctx.get::().is_some()); + } + + #[derive(Clone, Debug, Eq, PartialEq)] + struct StackHandle { + child: ChildHandle, + sibling: SiblingHandle, + } + + struct StackApp; + + #[async_trait] + impl AppDeployment for StackApp { + type Handle = StackHandle; + + async fn deploy(self, ctx: &mut DeployContext) -> Result { + let child = ctx.deploy_and_expose(ChildApp).await?; + let sibling = ctx.deploy_and_expose(SiblingApp).await?; + let stack = StackHandle { child, sibling }; + + ctx.expose(stack.clone())?; + + Ok(stack) + } + } + + #[tokio::test] + async fn nested_deployment_can_expose_multiple_typed_handles() { + let mut ctx = test_context(); + let stack = ctx.deploy(StackApp).await.expect("deploy stack app"); + + assert_eq!(stack.child, ChildHandle { id: "child" }); + assert_eq!(stack.sibling, SiblingHandle { id: "sibling" }); + assert!(ctx.get::().is_some()); + assert!(ctx.get::().is_some()); + assert!(ctx.get::().is_some()); + } + + fn test_context() -> DeployContext { + DeployContext::new(TestDeployment, NodeClients::default()) + } +} diff --git a/testing-framework/app/src/deployment.rs b/testing-framework/app/src/deployment.rs new file mode 100644 index 0000000..c350097 --- /dev/null +++ b/testing-framework/app/src/deployment.rs @@ -0,0 +1,30 @@ +use async_trait::async_trait; +use testing_framework_core::scenario::{Application, DynError}; + +use crate::DeployContext; + +/// Typed runtime capability returned by an application deployment. +/// +/// Handles are cloned when retrieved from the runtime registry. Managed +/// resources must be acquired through TF adapters, which register scenario +/// cleanup independently from these clones. +pub trait AppHandle: Clone + Send + Sync + 'static {} + +impl AppHandle for T where T: Clone + Send + Sync + 'static {} + +/// Deploys one reusable application preset and returns its typed handle. +#[async_trait] +pub trait AppDeployment: Send + 'static +where + E: Application, +{ + /// Runtime capability produced by this deployment. + type Handle: AppHandle; + + /// Prepares the application and returns its typed runtime access handle. + /// + /// Child deployments can be composed with [`DeployContext::deploy`] or + /// [`DeployContext::deploy_and_expose`]. Any handle exposed through the + /// context remains available through the scenario runtime. + async fn deploy(self, ctx: &mut DeployContext) -> Result; +} diff --git a/testing-framework/app/src/error.rs b/testing-framework/app/src/error.rs new file mode 100644 index 0000000..deca0ef --- /dev/null +++ b/testing-framework/app/src/error.rs @@ -0,0 +1,29 @@ +use thiserror::Error; + +#[derive(Debug, Error)] +/// Error returned when a typed application handle cannot be registered or +/// found. +pub enum AppDeployError { + /// No handle exists for the requested type and optional name. + #[error("app handle is not exposed: {type_name}{suffix}", suffix = format_name(.name))] + HandleMissing { + /// Rust type name of the requested handle. + type_name: &'static str, + /// Instance name, or `None` for the default unnamed handle. + name: Option, + }, + /// A handle already exists for the same type and optional name. + #[error("app handle is already exposed: {type_name}{suffix}", suffix = format_name(.name))] + DuplicateHandle { + /// Rust type name of the duplicate handle. + type_name: &'static str, + /// Instance name, or `None` for the default unnamed handle. + name: Option, + }, +} + +fn format_name(name: &Option) -> String { + name.as_ref() + .map(|name| format!(" named {name:?}")) + .unwrap_or_default() +} diff --git a/testing-framework/app/src/extension.rs b/testing-framework/app/src/extension.rs new file mode 100644 index 0000000..eac0d43 --- /dev/null +++ b/testing-framework/app/src/extension.rs @@ -0,0 +1,233 @@ +use async_trait::async_trait; +use testing_framework_core::scenario::{ + Application, CoreBuilderExt, DynError, NodeClients, PreparedRuntimeExtension, RunContext, + RuntimeExtensionFactory, internal::CoreBuilderAccess, +}; + +use crate::{AppDeployment, AppHandle, AppRuntime, DeployContext}; + +/// Adapts an [`AppDeployment`] to the scenario runtime extension lifecycle. +/// +/// A factory is normally installed through [`AppScenarioBuilderExt::with_app`] +/// rather than constructed directly. +pub struct AppDeploymentFactory { + app: A, +} + +impl AppDeploymentFactory { + /// Creates a runtime extension factory for `app`. + pub const fn new(app: A) -> Self { + Self { app } + } +} + +#[async_trait] +impl RuntimeExtensionFactory for AppDeploymentFactory +where + E: Application, + A: AppDeployment + Clone + Sync, +{ + async fn prepare( + &self, + deployment: &E::Deployment, + node_clients: NodeClients, + ) -> Result { + let mut ctx = DeployContext::new(deployment.clone(), node_clients); + + let handle = ctx.deploy(self.app.clone()).await?; + + if !ctx.contains::() { + ctx.expose(handle)?; + } + + let (handles, cleanup) = ctx.into_runtime_parts(); + let runtime = AppRuntime::new(handles); + + Ok(match cleanup { + Some(cleanup) => PreparedRuntimeExtension::with_cleanup(runtime, cleanup), + None => PreparedRuntimeExtension::new(runtime), + }) + } +} + +/// Adds composable application deployments to scenario builders. +pub trait AppScenarioBuilderExt: CoreBuilderAccess + Sized { + /// Registers an application deployment to prepare before workloads start. + /// + /// If the root deployment does not expose its returned handle itself, the + /// factory exposes it as the default handle automatically. + #[must_use] + fn with_app(self, app: A) -> Self + where + A: AppDeployment + Clone + Sync, + { + self.with_runtime_extension_factory(Box::new(AppDeploymentFactory::new(app))) + } +} + +impl AppScenarioBuilderExt for T where T: CoreBuilderAccess {} + +/// Retrieves application handles from a running scenario. +pub trait AppRunContextExt { + /// Returns the default handle for `T`, or `None` if it is not exposed. + fn app(&self) -> Option + where + T: AppHandle; + + /// Returns a named handle for `T`, or `None` if it is not exposed. + fn app_named(&self, name: &str) -> Option + where + T: AppHandle; + + /// Returns the default handle for `T` or an error when it is unavailable. + fn require_app(&self) -> Result + where + T: AppHandle; + + /// Returns a named handle for `T` or an error when it is unavailable. + fn require_app_named(&self, name: &str) -> Result + where + T: AppHandle; +} + +impl AppRunContextExt for RunContext +where + E: Application, +{ + fn app(&self) -> Option + where + T: AppHandle, + { + self.extension::()?.get() + } + + fn app_named(&self, name: &str) -> Option + where + T: AppHandle, + { + self.extension::()?.get_named(name) + } + + fn require_app(&self) -> Result + where + T: AppHandle, + { + self.require_extension::()? + .require() + .map_err(Into::into) + } + + fn require_app_named(&self, name: &str) -> Result + where + T: AppHandle, + { + self.require_extension::()? + .require_named(name) + .map_err(Into::into) + } +} + +#[cfg(test)] +mod tests { + use std::sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + }; + + use async_trait::async_trait; + use testing_framework_core::{ + scenario::{Application, DynError, NodeClients, RuntimeExtensionFactory}, + topology::DeploymentDescriptor, + }; + + use super::AppDeploymentFactory; + use crate::{AppDeployment, DeployContext}; + + #[derive(Clone)] + struct TestDeployment; + + impl DeploymentDescriptor for TestDeployment { + fn node_count(&self) -> usize { + 0 + } + } + + struct TestEnv; + + #[async_trait] + impl Application for TestEnv { + type Deployment = TestDeployment; + type NodeClient = (); + type NodeConfig = (); + } + + #[derive(Clone)] + struct FailingApp { + dropped: Arc, + } + + #[derive(Clone)] + struct OwnedHandle { + _resource: Arc, + } + + struct DropProbe { + dropped: Arc, + } + + impl Drop for DropProbe { + fn drop(&mut self) { + self.dropped.store(true, Ordering::SeqCst); + } + } + + #[async_trait] + impl AppDeployment for FailingApp { + type Handle = OwnedHandle; + + async fn deploy(self, ctx: &mut DeployContext) -> Result { + ctx.expose(OwnedHandle { + _resource: Arc::new(DropProbe { + dropped: Arc::clone(&self.dropped), + }), + })?; + Err("deployment failed".into()) + } + } + + #[tokio::test] + async fn deployment_failure_cleans_up_partial_resources() { + let dropped = Arc::new(AtomicBool::new(false)); + let factory = AppDeploymentFactory::new(FailingApp { + dropped: Arc::clone(&dropped), + }); + + let result = factory + .prepare(test_deployment(), NodeClients::default()) + .await; + + assert!(result.is_err(), "deployment must fail"); + + assert!(dropped.load(Ordering::SeqCst)); + } + + #[tokio::test] + async fn app_factory_can_prepare_more_than_once() { + let dropped = Arc::new(AtomicBool::new(false)); + let factory = AppDeploymentFactory::new(FailingApp { + dropped: Arc::clone(&dropped), + }); + + for _ in 0..2 { + let result = factory + .prepare(test_deployment(), NodeClients::default()) + .await; + assert!(result.is_err(), "deployment must fail"); + } + } + + fn test_deployment() -> &'static TestDeployment { + static DEPLOYMENT: TestDeployment = TestDeployment; + &DEPLOYMENT + } +} diff --git a/testing-framework/app/src/host.rs b/testing-framework/app/src/host.rs new file mode 100644 index 0000000..1caa38f --- /dev/null +++ b/testing-framework/app/src/host.rs @@ -0,0 +1,55 @@ +use async_trait::async_trait; +use testing_framework_core::{ + scenario::{Application, DynError, NodeAccess, ScenarioBuilder}, + topology::DeploymentDescriptor, +}; +use testing_framework_runner_local::{LocalDeployerEnv, ProcessDeployer}; + +#[derive(Clone, Default)] +/// Empty outer topology for scenarios whose entire system is deployed as apps. +/// +/// Application deployments registered with [`crate::AppScenarioBuilderExt`] +/// provide all processes and clusters, so this topology contains no nodes. +pub struct AppHostTopology; + +impl DeploymentDescriptor for AppHostTopology { + fn node_count(&self) -> usize { + 0 + } +} + +/// Testing-framework application environment for [`AppHost`] scenarios. +/// +/// This environment intentionally has no outer node client. +/// Application-specific clients are exposed as typed handles instead. +pub struct AppHostEnv; + +#[async_trait] +impl Application for AppHostEnv { + type Deployment = AppHostTopology; + type NodeClient = (); + type NodeConfig = (); + + fn build_node_client(_access: &NodeAccess) -> Result { + Err(std::io::Error::other("app host does not expose node clients").into()) + } +} + +#[async_trait] +impl LocalDeployerEnv for AppHostEnv {} + +/// Entry point for a scenario composed entirely from application deployments. +pub struct AppHost; + +impl AppHost { + /// Creates an empty scenario builder ready for `.with_app(...)` calls. + #[must_use] + pub fn scenario() -> AppHostScenarioBuilder { + ScenarioBuilder::with_deployment(AppHostTopology) + } +} + +/// Scenario builder for an application-hosted heterogeneous stack. +pub type AppHostScenarioBuilder = ScenarioBuilder; +/// Local process deployer used to execute an [`AppHost`] scenario. +pub type AppHostLocalDeployer = ProcessDeployer; diff --git a/testing-framework/app/src/lib.rs b/testing-framework/app/src/lib.rs new file mode 100644 index 0000000..b80fed2 --- /dev/null +++ b/testing-framework/app/src/lib.rs @@ -0,0 +1,35 @@ +//! Composable application deployment for heterogeneous test systems. +//! +//! The core testing framework models one +//! [`Application`](testing_framework_core::scenario::Application) +//! and its uniform node topology. This crate adds an application layer for +//! scenarios that also need singleton processes, additional clusters, +//! attached services, or a stack of several different applications. +//! +//! Implement [`AppDeployment`] in the application repository, compose child +//! deployments through [`DeployContext`], and expose typed handles to workloads +//! with [`AppRunContextExt`]. TF adapters register managed resources with the +//! scenario cleanup path, while attached and external apps remain explicit. + +#![warn(missing_docs)] + +mod cleanup; +mod context; +mod deployment; +mod error; +mod extension; +mod host; +mod local; +mod process; +mod registry; + +pub use context::DeployContext; +pub use deployment::{AppDeployment, AppHandle}; +pub use error::AppDeployError; +pub use extension::{AppDeploymentFactory, AppRunContextExt, AppScenarioBuilderExt}; +pub use host::{ + AppHost, AppHostEnv, AppHostLocalDeployer, AppHostScenarioBuilder, AppHostTopology, +}; +pub use local::LocalAppCluster; +pub use process::{LocalProcessApp, LocalProcessHandle}; +pub use registry::{AppRuntime, HandleRegistry}; diff --git a/testing-framework/app/src/local.rs b/testing-framework/app/src/local.rs new file mode 100644 index 0000000..d3c11c9 --- /dev/null +++ b/testing-framework/app/src/local.rs @@ -0,0 +1,238 @@ +use std::sync::{ + Arc, + atomic::{AtomicBool, Ordering}, +}; + +use testing_framework_core::{ + scenario::{DynError, NodeClients, StartNodeOptions, StartedNode, internal::CleanupGuard}, + topology::DeploymentDescriptor, +}; +use testing_framework_runner_local::{LocalDeployerEnv, ManualCluster, ProcessDeployer}; + +/// Control handle for an additional uniform cluster of local processes. +/// +/// Unlike the outer scenario cluster, this cluster is deployed as one child of +/// a heterogeneous application stack. The scenario owns its lifetime and stops +/// every node during cleanup even if a handle clone remains elsewhere. +pub struct LocalAppCluster +where + E: LocalDeployerEnv, +{ + deployment: E::Deployment, + owner: Arc>, +} + +struct LocalClusterOwner +where + E: LocalDeployerEnv, +{ + cluster: ManualCluster, + closed: AtomicBool, +} + +impl Drop for LocalClusterOwner +where + E: LocalDeployerEnv, +{ + fn drop(&mut self) { + self.close(); + } +} + +impl LocalClusterOwner +where + E: LocalDeployerEnv, +{ + fn close(&self) { + if !self.closed.swap(true, Ordering::AcqRel) { + self.cluster.stop_all(); + } + } + + fn ensure_open(&self) -> Result<(), DynError> { + if self.closed.load(Ordering::Acquire) { + Err("local app cluster is no longer owned by an active run".into()) + } else { + Ok(()) + } + } +} + +struct LocalClusterCleanup +where + E: LocalDeployerEnv, +{ + owner: Arc>, +} + +impl CleanupGuard for LocalClusterCleanup +where + E: LocalDeployerEnv, +{ + fn cleanup(self: Box) { + self.owner.close(); + } +} + +impl Clone for LocalAppCluster +where + E: LocalDeployerEnv, +{ + fn clone(&self) -> Self { + Self { + deployment: self.deployment.clone(), + owner: Arc::clone(&self.owner), + } + } +} + +impl LocalAppCluster +where + E: LocalDeployerEnv, +{ + /// Starts every node described by `deployment` and waits for readiness. + pub(crate) async fn start(deployment: E::Deployment) -> Result { + let cluster = + ProcessDeployer::::default().manual_cluster_from_descriptors(deployment.clone()); + let owner = Arc::new(LocalClusterOwner { + cluster, + closed: AtomicBool::new(false), + }); + + start_all_nodes(&owner.cluster, deployment.node_count()).await?; + owner.cluster.wait_network_ready().await?; + + Ok(Self { deployment, owner }) + } + + /// Returns this cluster's deployment descriptor. + #[must_use] + pub fn deployment(&self) -> &E::Deployment { + &self.deployment + } + + /// Returns the number of nodes described by the deployment. + #[must_use] + pub fn node_count(&self) -> usize { + self.deployment.node_count() + } + + /// Returns a shared snapshot-capable collection of node clients. + #[must_use] + pub fn node_clients(&self) -> NodeClients { + self.owner.cluster.node_clients() + } + + /// Returns the client for `name`, if that node has started. + #[must_use] + pub fn node_client(&self, name: &str) -> Option { + self.owner.cluster.node_client(name) + } + + /// Returns the operating-system process ID for `name`, if it is running. + #[must_use] + pub fn node_pid(&self, name: &str) -> Option { + self.owner.cluster.node_pid(name) + } + + /// Returns a snapshot of all currently available node clients. + #[must_use] + pub fn clients(&self) -> Vec { + self.node_clients().snapshot() + } + + /// Returns the first available node client. + #[must_use] + pub fn first_client(&self) -> Option { + self.node_clients() + .with_clients(|clients| clients.first().cloned()) + } + + /// Starts the named node with default start options. + pub async fn start_node(&self, name: &str) -> Result, DynError> { + self.owner.ensure_open()?; + Ok(self.owner.cluster.start_node(name).await?) + } + + /// Starts the named node with explicit start options. + pub async fn start_node_with( + &self, + name: &str, + options: StartNodeOptions, + ) -> Result, DynError> { + self.owner.ensure_open()?; + Ok(self.owner.cluster.start_node_with(name, options).await?) + } + + /// Stops the named node. + pub async fn stop_node(&self, name: &str) -> Result<(), DynError> { + self.owner.ensure_open()?; + Ok(self.owner.cluster.stop_node(name).await?) + } + + /// Restarts the named node with its existing configuration. + pub async fn restart_node(&self, name: &str) -> Result<(), DynError> { + self.owner.ensure_open()?; + Ok(self.owner.cluster.restart_node(name).await?) + } + + /// Restarts the named node with explicit start options. + pub async fn restart_node_with( + &self, + name: &str, + options: StartNodeOptions, + ) -> Result<(), DynError> { + self.owner.ensure_open()?; + Ok(self.owner.cluster.restart_node_with(name, options).await?) + } + + /// Waits until the cluster-level network readiness condition succeeds. + pub async fn wait_network_ready(&self) -> Result<(), DynError> { + self.owner.ensure_open()?; + Ok(self.owner.cluster.wait_network_ready().await?) + } + + /// Waits until the named node reports ready. + pub async fn wait_node_ready(&self, name: &str) -> Result<(), DynError> { + self.owner.ensure_open()?; + Ok(self.owner.cluster.wait_node_ready(name).await?) + } + + /// Stops every node while keeping the cluster available for a later start. + pub fn stop_all(&self) -> Result<(), DynError> { + self.owner.ensure_open()?; + self.owner.cluster.stop_all(); + Ok(()) + } + + /// Starts every node with its original deployment and waits for readiness. + pub async fn start_all(&self) -> Result<(), DynError> { + self.owner.ensure_open()?; + start_all_nodes(&self.owner.cluster, self.node_count()).await?; + self.owner.cluster.wait_network_ready().await?; + Ok(()) + } + + /// Stops and starts every node, then waits for network readiness. + pub async fn restart_all(&self) -> Result<(), DynError> { + self.stop_all()?; + self.start_all().await + } + + pub(crate) fn cleanup_guard(&self) -> Box { + Box::new(LocalClusterCleanup { + owner: Arc::clone(&self.owner), + }) + } +} + +async fn start_all_nodes(cluster: &ManualCluster, node_count: usize) -> Result<(), DynError> +where + E: LocalDeployerEnv, +{ + for index in 0..node_count { + cluster.start_node(&format!("node-{index}")).await?; + } + + Ok(()) +} diff --git a/testing-framework/app/src/process.rs b/testing-framework/app/src/process.rs new file mode 100644 index 0000000..99e4346 --- /dev/null +++ b/testing-framework/app/src/process.rs @@ -0,0 +1,387 @@ +use std::{ + future::Future, + path::PathBuf, + pin::Pin, + sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + }, +}; + +use async_trait::async_trait; +use testing_framework_core::scenario::{Application, DynError, internal::CleanupGuard}; +use testing_framework_runner_local::{LaunchSpec, NodeEndpoints, ProcessNode}; +use tokio::sync::Mutex; + +use crate::{AppDeployment, DeployContext}; + +type ReadinessFuture = Pin> + Send>>; +type Readiness = Arc ReadinessFuture + Send + Sync>; + +/// One local binary process that can be composed with other app deployments. +/// +/// This is the small path for heterogeneous stacks. It does not require the +/// process to pretend to be a uniform TF node cluster. The app repository owns +/// its launch files, client type, and readiness check; TF owns process +/// lifetime, working directories, and teardown. +pub struct LocalProcessApp { + label: String, + launch: LaunchSpec, + endpoints: NodeEndpoints, + client: C, + readiness: Option>, + keep_tempdir: bool, + persist_dir: Option, + snapshot_dir: Option, +} + +impl LocalProcessApp +where + C: Clone + Send + Sync + 'static, +{ + /// Describes a process and the client returned through its runtime handle. + /// + /// `endpoints` describes the addresses assigned by the application-specific + /// launch configuration; the generic process layer does not allocate them. + #[must_use] + pub fn new( + label: impl Into, + launch: LaunchSpec, + endpoints: NodeEndpoints, + client: C, + ) -> Self { + Self { + label: label.into(), + launch, + endpoints, + client, + readiness: None, + keep_tempdir: false, + persist_dir: None, + snapshot_dir: None, + } + } + + /// Adds an asynchronous readiness check that runs after the process starts. + /// + /// A readiness failure stops the process and fails deployment. + #[must_use] + pub fn with_readiness(mut self, readiness: F) -> Self + where + F: Fn(NodeEndpoints, C) -> Fut + Send + Sync + 'static, + Fut: Future> + Send + 'static, + { + self.readiness = Some(Arc::new(move |endpoints, client| { + Box::pin(readiness(endpoints, client)) + })); + self + } + + /// Controls whether the generated process working directory survives + /// teardown. + #[must_use] + pub fn keep_tempdir(mut self, keep: bool) -> Self { + self.keep_tempdir = keep; + self + } + + /// Copies persisted state from `path` into the process working directory. + #[must_use] + pub fn with_persist_dir(mut self, path: impl Into) -> Self { + self.persist_dir = Some(path.into()); + self + } + + /// Copies snapshot state from `path` into the process working directory. + #[must_use] + pub fn with_snapshot_dir(mut self, path: impl Into) -> Self { + self.snapshot_dir = Some(path.into()); + self + } +} + +#[async_trait] +impl AppDeployment for LocalProcessApp +where + E: Application, + C: Clone + Send + Sync + 'static, +{ + type Handle = LocalProcessHandle; + + async fn deploy(self, ctx: &mut DeployContext) -> Result { + let Self { + label, + launch, + endpoints, + client, + readiness, + keep_tempdir, + persist_dir, + snapshot_dir, + } = self; + + let mut process = ProcessNode::spawn( + &label, + (), + move |(), _working_dir, _label| Ok(launch), + move |()| Ok(endpoints), + keep_tempdir, + persist_dir.as_deref(), + snapshot_dir.as_deref(), + move |_endpoints| Ok(client), + ) + .await?; + + let endpoints = process.endpoints().clone(); + let client = process.client(); + + if let Some(readiness) = &readiness { + if let Err(error) = readiness(endpoints.clone(), client.clone()).await { + process.stop().await; + return Err(error); + } + } + + let state = Arc::new(LocalProcessState { + process: Mutex::new(process), + endpoints: endpoints.clone(), + client: client.clone(), + readiness, + closed: AtomicBool::new(false), + }); + let handle = LocalProcessHandle { + state, + endpoints, + client, + }; + ctx.register_cleanup(handle.cleanup_guard()); + + Ok(handle) + } +} + +struct LocalProcessState +where + C: Clone + Send + Sync + 'static, +{ + process: Mutex>, + endpoints: NodeEndpoints, + client: C, + readiness: Option>, + closed: AtomicBool, +} + +impl LocalProcessState +where + C: Clone + Send + Sync + 'static, +{ + fn ensure_open(&self) -> Result<(), DynError> { + if self.closed.load(Ordering::Acquire) { + Err("local process is no longer owned by an active run".into()) + } else { + Ok(()) + } + } + + fn close(&self) { + if self.closed.swap(true, Ordering::AcqRel) { + return; + } + + if let Ok(mut process) = self.process.try_lock() { + process.stop_blocking(); + return; + } + + std::thread::scope(|scope| { + let worker = scope.spawn(|| self.process.blocking_lock().stop_blocking()); + let _ = worker.join(); + }); + } + + async fn wait_ready(&self) -> Result<(), DynError> { + if let Some(readiness) = &self.readiness { + readiness(self.endpoints.clone(), self.client.clone()).await?; + } + Ok(()) + } +} + +struct LocalProcessCleanup +where + C: Clone + Send + Sync + 'static, +{ + state: Arc>, +} + +impl CleanupGuard for LocalProcessCleanup +where + C: Clone + Send + Sync + 'static, +{ + fn cleanup(self: Box) { + self.state.close(); + } +} + +/// Shared ownership and control handle for one local application process. +/// +/// Clones refer to the same process. The scenario owns its lifetime, so cleanup +/// stops it even if a handle clone remains elsewhere. +pub struct LocalProcessHandle +where + C: Clone + Send + Sync + 'static, +{ + state: Arc>, + endpoints: NodeEndpoints, + client: C, +} + +impl Clone for LocalProcessHandle +where + C: Clone + Send + Sync + 'static, +{ + fn clone(&self) -> Self { + Self { + state: Arc::clone(&self.state), + endpoints: self.endpoints.clone(), + client: self.client.clone(), + } + } +} + +impl LocalProcessHandle +where + C: Clone + Send + Sync + 'static, +{ + /// Returns a clone of the application-specific client. + #[must_use] + pub fn client(&self) -> C { + self.client.clone() + } + + /// Returns the process endpoints supplied at deployment time. + #[must_use] + pub fn endpoints(&self) -> &NodeEndpoints { + &self.endpoints + } + + /// Returns the operating-system process ID. + pub async fn pid(&self) -> u32 { + self.state.process.lock().await.pid() + } + + /// Returns whether the child process is still running. + pub async fn is_running(&self) -> bool { + if self.state.closed.load(Ordering::Acquire) { + return false; + } + self.state.process.lock().await.is_running() + } + + /// Returns the generated working directory used by the process. + pub async fn working_dir(&self) -> PathBuf { + self.state.process.lock().await.working_dir().to_owned() + } + + /// Starts a process that was previously stopped. + pub async fn start(&self) -> Result<(), DynError> { + self.state.ensure_open()?; + let mut process = self.state.process.lock().await; + if !process.is_running() { + process.restart().await?; + } + drop(process); + self.state.wait_ready().await?; + Ok(()) + } + + /// Restarts the process with its original launch specification. + pub async fn restart(&self) -> Result<(), DynError> { + self.state.ensure_open()?; + self.state.process.lock().await.restart().await?; + self.state.wait_ready().await?; + Ok(()) + } + + /// Stops the process without waiting for the handle to be dropped. + pub async fn stop(&self) -> Result<(), DynError> { + self.state.ensure_open()?; + self.state.process.lock().await.stop().await; + Ok(()) + } + + /// Prevents deletion of the temporary working directory on teardown. + pub async fn keep_tempdir(&self) -> std::io::Result<()> { + self.state.process.lock().await.keep_tempdir() + } + + pub(crate) fn cleanup_guard(&self) -> Box { + Box::new(LocalProcessCleanup { + state: Arc::clone(&self.state), + }) + } +} + +#[cfg(all(test, unix))] +mod tests { + use async_trait::async_trait; + use testing_framework_core::{ + scenario::{Application, NodeClients}, + topology::DeploymentDescriptor, + }; + use testing_framework_runner_local::{LaunchSpec, NodeEndpoints}; + + use super::LocalProcessApp; + use crate::DeployContext; + + #[derive(Clone)] + struct TestDeployment; + + impl DeploymentDescriptor for TestDeployment { + fn node_count(&self) -> usize { + 0 + } + } + + struct TestEnv; + + #[async_trait] + impl Application for TestEnv { + type Deployment = TestDeployment; + type NodeClient = (); + type NodeConfig = (); + } + + #[tokio::test] + async fn process_lifetime_follows_scenario_ownership() { + let launch = LaunchSpec { + binary: "/bin/sh".into(), + args: vec!["-c".into(), "sleep 30".into()], + ..LaunchSpec::default() + }; + let app = LocalProcessApp::new( + "test-process", + launch, + NodeEndpoints::default(), + "client".to_owned(), + ) + .with_readiness(|_endpoints, client| async move { + assert_eq!(client, "client"); + Ok(()) + }); + let mut ctx = DeployContext::::new(TestDeployment, NodeClients::default()); + + let handle = ctx.deploy(app).await.expect("deploy process"); + assert!(handle.is_running().await); + assert_eq!(handle.client(), "client"); + + handle.stop().await.unwrap(); + assert!(!handle.is_running().await); + handle.start().await.unwrap(); + assert!(handle.is_running().await); + + drop(ctx); + assert!(!handle.is_running().await); + assert!(handle.restart().await.is_err()); + } +} diff --git a/testing-framework/app/src/registry.rs b/testing-framework/app/src/registry.rs new file mode 100644 index 0000000..51bb57b --- /dev/null +++ b/testing-framework/app/src/registry.rs @@ -0,0 +1,281 @@ +use std::{ + any::{Any, TypeId, type_name}, + sync::Arc, +}; + +use crate::{AppDeployError, AppHandle}; + +const DEFAULT_HANDLE_NAME: &str = ""; + +#[derive(Default)] +/// Type-indexed storage for application runtime handles. +/// +/// One unnamed handle may be stored per concrete type. Named handles allow +/// multiple instances of the same type. Entries retain exposure order and are +/// released in reverse order. Managed resource teardown is registered +/// separately by TF adapters and does not depend on handle clone counts. +pub struct HandleRegistry { + handles: Vec<(HandleKey, Box)>, +} + +impl HandleRegistry { + /// Creates an empty registry. + #[must_use] + pub fn new() -> Self { + Self::default() + } + + /// Stores the default unnamed handle for `T`. + /// + /// Returns [`AppDeployError::DuplicateHandle`] if one is already stored. + pub fn expose(&mut self, handle: T) -> Result<(), AppDeployError> + where + T: AppHandle, + { + self.expose_named(DEFAULT_HANDLE_NAME, handle) + } + + /// Stores `handle` under its concrete type and `name`. + /// + /// Returns [`AppDeployError::DuplicateHandle`] for an existing type/name + /// pair. + pub fn expose_named( + &mut self, + name: impl Into, + handle: T, + ) -> Result<(), AppDeployError> + where + T: AppHandle, + { + let key = HandleKey::new::(name); + if self.handles.iter().any(|(existing, _)| existing == &key) { + return Err(AppDeployError::DuplicateHandle { + type_name: type_name::(), + name: key.display_name().map(ToOwned::to_owned), + }); + } + + self.handles.push((key, Box::new(handle))); + Ok(()) + } + + /// Returns a clone of the default handle for `T`, if present. + #[must_use] + pub fn get(&self) -> Option + where + T: AppHandle, + { + self.get_named(DEFAULT_HANDLE_NAME) + } + + /// Returns a clone of the named handle for `T`, if present. + #[must_use] + pub fn get_named(&self, name: &str) -> Option + where + T: AppHandle, + { + let key = HandleKey::new::(name); + self.handles + .iter() + .find(|(existing, _)| existing == &key) + .and_then(|(_, handle)| handle.as_ref().downcast_ref::()) + .cloned() + } + + /// Returns the default handle for `T` or [`AppDeployError::HandleMissing`]. + pub fn require(&self) -> Result + where + T: AppHandle, + { + self.require_named(DEFAULT_HANDLE_NAME) + } + + /// Returns the named handle for `T` or [`AppDeployError::HandleMissing`]. + pub fn require_named(&self, name: &str) -> Result + where + T: AppHandle, + { + self.get_named(name) + .ok_or_else(|| AppDeployError::HandleMissing { + type_name: type_name::(), + name: (!name.is_empty()).then(|| name.to_owned()), + }) + } + + /// Returns whether the default handle for `T` is present. + #[must_use] + pub fn contains(&self) -> bool + where + T: AppHandle, + { + self.contains_named::(DEFAULT_HANDLE_NAME) + } + + /// Returns whether the named handle for `T` is present. + #[must_use] + pub fn contains_named(&self, name: &str) -> bool + where + T: AppHandle, + { + let key = HandleKey::new::(name); + self.handles.iter().any(|(existing, _)| existing == &key) + } + + /// Returns whether the registry contains no handles. + #[must_use] + pub fn is_empty(&self) -> bool { + self.handles.is_empty() + } +} + +impl Drop for HandleRegistry { + fn drop(&mut self) { + // Release arbitrary handle state deterministically. Managed resources + // use the deployment cleanup stack, whose order follows acquisition. + while self.handles.pop().is_some() {} + } +} + +#[derive(Clone, Default)] +/// Scenario runtime extension containing exposed application handles. +/// +/// Runtime clones share one registry. Managed resources have a separate +/// scenario-owned cleanup stack. +pub struct AppRuntime { + handles: Arc, +} + +impl AppRuntime { + /// Wraps a prepared handle registry as a scenario runtime extension. + #[must_use] + pub fn new(handles: HandleRegistry) -> Self { + Self { + handles: Arc::new(handles), + } + } + + /// Returns a clone of the default handle for `T`, if present. + #[must_use] + pub fn get(&self) -> Option + where + T: AppHandle, + { + self.handles.get() + } + + /// Returns a clone of the named handle for `T`, if present. + #[must_use] + pub fn get_named(&self, name: &str) -> Option + where + T: AppHandle, + { + self.handles.get_named(name) + } + + /// Returns the default handle for `T` or [`AppDeployError::HandleMissing`]. + pub fn require(&self) -> Result + where + T: AppHandle, + { + self.handles.require() + } + + /// Returns the named handle for `T` or [`AppDeployError::HandleMissing`]. + pub fn require_named(&self, name: &str) -> Result + where + T: AppHandle, + { + self.handles.require_named(name) + } +} + +#[derive(Eq, Hash, PartialEq)] +struct HandleKey { + type_id: TypeId, + name: String, +} + +impl HandleKey { + fn new(name: impl Into) -> Self { + Self { + type_id: TypeId::of::(), + name: name.into(), + } + } + + fn display_name(&self) -> Option<&str> { + (!self.name.is_empty()).then_some(self.name.as_str()) + } +} + +type StoredHandle = dyn Any + Send + Sync; + +#[cfg(test)] +mod tests { + use std::sync::{Arc, Mutex}; + + use super::HandleRegistry; + + #[derive(Clone, Debug, Eq, PartialEq)] + struct Handle(u8); + + #[test] + fn named_handles_allow_multiple_instances_of_one_type() { + let mut handles = HandleRegistry::new(); + handles.expose_named("left", Handle(1)).unwrap(); + handles.expose_named("right", Handle(2)).unwrap(); + + assert_eq!(handles.get_named("left"), Some(Handle(1))); + assert_eq!(handles.get_named("right"), Some(Handle(2))); + } + + #[test] + fn duplicate_handle_is_rejected() { + let mut handles = HandleRegistry::new(); + handles.expose(Handle(1)).unwrap(); + + let error = handles.expose(Handle(2)).unwrap_err(); + assert!(error.to_string().contains("already exposed")); + assert_eq!(handles.get(), Some(Handle(1))); + } + + #[test] + fn handles_drop_in_reverse_exposure_order() { + let order = Arc::new(Mutex::new(Vec::new())); + let mut handles = HandleRegistry::new(); + handles + .expose_named("first", OwnedHandle::new("first", Arc::clone(&order))) + .unwrap(); + handles + .expose_named("second", OwnedHandle::new("second", Arc::clone(&order))) + .unwrap(); + + drop(handles); + + assert_eq!(*order.lock().unwrap(), ["second", "first"]); + } + + #[derive(Clone)] + struct OwnedHandle { + _resource: Arc, + } + + impl OwnedHandle { + fn new(label: &'static str, order: Arc>>) -> Self { + Self { + _resource: Arc::new(OwnedResource { label, order }), + } + } + } + + struct OwnedResource { + label: &'static str, + order: Arc>>, + } + + impl Drop for OwnedResource { + fn drop(&mut self) { + self.order.lock().unwrap().push(self.label); + } + } +} diff --git a/testing-framework/core/src/scenario/runtime/extensions.rs b/testing-framework/core/src/scenario/runtime/extensions.rs index 1b9f9f4..a4e5aa3 100644 --- a/testing-framework/core/src/scenario/runtime/extensions.rs +++ b/testing-framework/core/src/scenario/runtime/extensions.rs @@ -84,6 +84,11 @@ impl RuntimeExtensions { .and_then(|value| value.downcast_ref::()) .cloned() } + + #[must_use] + pub fn is_empty(&self) -> bool { + self.values.is_empty() + } } #[derive(Default)] diff --git a/testing-framework/deployers/local/src/binary/mod.rs b/testing-framework/deployers/local/src/binary/mod.rs index 24f326a..d8111f6 100644 --- a/testing-framework/deployers/local/src/binary/mod.rs +++ b/testing-framework/deployers/local/src/binary/mod.rs @@ -17,8 +17,9 @@ use cache::BinaryCache; pub(super) use types::optional_path_display; pub use types::{ BinaryProviderError, BinaryProviderRef, BuildBinaryProvider, BuildCommand, - DownloadBinaryProvider, DownloadChecksum, DownloadUrl, EnvBinaryProvider, - FallbackBinaryProvider, PathBinaryProvider, + DownloadBinaryProvider, DownloadChecksum, DownloadProcessor, DownloadProcessorError, + DownloadProcessorFn, DownloadUrl, EnvBinaryProvider, FallbackBinaryProvider, + PathBinaryProvider, }; /// Resolves an executable path for a local process. diff --git a/testing-framework/deployers/local/src/binary/providers/download.rs b/testing-framework/deployers/local/src/binary/providers/download.rs index 24a4882..c24825c 100644 --- a/testing-framework/deployers/local/src/binary/providers/download.rs +++ b/testing-framework/deployers/local/src/binary/providers/download.rs @@ -17,7 +17,7 @@ use sha2::{Digest as _, Sha256}; use tracing::info; use crate::binary::{ - BinaryProvider, BinaryProviderError, DownloadBinaryProvider, DownloadUrl, + BinaryProvider, BinaryProviderError, DownloadBinaryProvider, DownloadChecksum, DownloadUrl, lock::BinaryProviderLock, optional_path_display, }; @@ -33,7 +33,7 @@ impl BinaryProvider for DownloadBinaryProvider { let bytes = self.download_bytes(&url)?; self.verify_checksum(&path, &bytes)?; - self.write_binary(&path, &bytes)?; + self.prepare_binary(&path, &bytes)?; Ok(Some(path)) } @@ -44,8 +44,14 @@ impl BinaryProvider for DownloadBinaryProvider { fn cache_key(&self) -> String { format!( - "download:{}:{}", + "download:{}:{}:{}:{}", self.url.cache_key(), + self.sha256 + .as_ref() + .map_or_else(String::new, DownloadChecksum::cache_key), + self.processor + .as_ref() + .map_or("", |processor| processor.cache_key()), optional_path_display(&self.cache_dir) ) } @@ -60,6 +66,15 @@ impl DownloadUrl { } } +impl DownloadChecksum { + fn cache_key(&self) -> String { + match self { + Self::Fixed(checksum) => checksum.to_ascii_lowercase(), + Self::Env(env_var) => format!("env:{env_var}"), + } + } +} + impl DownloadBinaryProvider { fn cached_binary_path(&self, url: &str) -> Result { let cache_dir = self.cache_dir(); @@ -92,13 +107,76 @@ impl DownloadBinaryProvider { }) } - fn write_binary(&self, path: &Path, bytes: &[u8]) -> Result<(), BinaryProviderError> { - fs::write(path, bytes).map_err(|source| BinaryProviderError::Io { - path: path.to_owned(), + fn prepare_binary(&self, path: &Path, bytes: &[u8]) -> Result<(), BinaryProviderError> { + let artifact = path.with_extension("download"); + let output = path.with_extension("part"); + self.remove_temporary_file(&artifact); + self.remove_temporary_file(&output); + + let result = self + .materialize_output(&artifact, &output, bytes) + .and_then(|()| { + self.ensure_processed_output(&output)?; + self.make_executable(&output)?; + fs::rename(&output, path).map_err(|source| BinaryProviderError::Io { + path: path.to_owned(), + source, + }) + }); + + self.remove_temporary_file(&artifact); + if result.is_err() { + self.remove_temporary_file(&output); + } + + result + } + + fn materialize_output( + &self, + artifact: &Path, + output: &Path, + bytes: &[u8], + ) -> Result<(), BinaryProviderError> { + let Some(processor) = &self.processor else { + return fs::write(output, bytes).map_err(|source| BinaryProviderError::Io { + path: output.to_owned(), + source, + }); + }; + + fs::write(artifact, bytes).map_err(|source| BinaryProviderError::Io { + path: artifact.to_owned(), source, })?; + processor.process(artifact, output).map_err(|source| { + BinaryProviderError::DownloadProcessing { + processor: processor.cache_key().to_owned(), + source, + } + }) + } - self.make_executable(path) + fn ensure_processed_output(&self, output: &Path) -> Result<(), BinaryProviderError> { + if output.is_file() { + return Ok(()); + } + + Err(BinaryProviderError::MissingProcessedOutput { + processor: self.processor.as_ref().map_or_else( + || "identity".to_owned(), + |processor| processor.cache_key().to_owned(), + ), + path: output.to_owned(), + }) + } + + fn remove_temporary_file(&self, path: &Path) { + if let Err(error) = fs::remove_file(path) + && error.kind() != std::io::ErrorKind::NotFound + { + tracing::warn!(path = %path.display(), %error, "failed to remove temporary download file"); + } } fn verify_checksum(&self, path: &Path, bytes: &[u8]) -> Result<(), BinaryProviderError> { @@ -163,6 +241,14 @@ impl DownloadBinaryProvider { fn download_file_name(&self, url: &str) -> String { let mut hasher = DefaultHasher::new(); url.hash(&mut hasher); + self.sha256 + .as_ref() + .and_then(DownloadChecksum::resolve) + .hash(&mut hasher); + self.processor + .as_ref() + .map(|processor| processor.cache_key()) + .hash(&mut hasher); format!("binary-{:x}", hasher.finish()) } diff --git a/testing-framework/deployers/local/src/binary/tests.rs b/testing-framework/deployers/local/src/binary/tests.rs index cb29dd4..15c1a26 100644 --- a/testing-framework/deployers/local/src/binary/tests.rs +++ b/testing-framework/deployers/local/src/binary/tests.rs @@ -158,6 +158,7 @@ fn downloads_binary_from_minimal_http_server() { url: DownloadUrl::Fixed(server.url()), sha256: Some(DownloadChecksum::Fixed(sha256_hex(body))), cache_dir: Some(temp.path().join("cache")), + processor: None, }; let path = provider.resolve().expect("download provider resolves"); @@ -172,6 +173,7 @@ fn rejects_download_checksum_mismatch() { url: DownloadUrl::Fixed(server.url()), sha256: Some(DownloadChecksum::Fixed("00".repeat(32))), cache_dir: Some(temp.path().join("cache")), + processor: None, }; let error = provider @@ -184,6 +186,61 @@ fn rejects_download_checksum_mismatch() { )); } +#[test] +fn processes_downloaded_artifact_before_publishing_binary() { + let temp = TempDir::new().expect("temp dir"); + let body = b"archive:downloaded-node"; + let server = SingleResponseServer::start(body); + let process_count = Arc::new(AtomicUsize::new(0)); + let callback_count = Arc::clone(&process_count); + let provider = DownloadBinaryProvider { + url: DownloadUrl::Fixed(server.url()), + sha256: Some(DownloadChecksum::Fixed(sha256_hex(body))), + cache_dir: Some(temp.path().join("cache")), + processor: None, + } + .with_processor_fn("strip-test-archive-v1", move |artifact, output| { + callback_count.fetch_add(1, Ordering::SeqCst); + let contents = fs::read(artifact)?; + fs::write( + output, + contents.strip_prefix(b"archive:").unwrap_or(&contents), + )?; + Ok(()) + }); + + let path = provider.resolve().expect("processed download resolves"); + + assert_eq!( + fs::read(path).expect("processed binary"), + b"downloaded-node" + ); + assert_eq!(process_count.load(Ordering::SeqCst), 1); +} + +#[test] +fn rejects_processor_that_does_not_create_output() { + let temp = TempDir::new().expect("temp dir"); + let body = b"archive"; + let server = SingleResponseServer::start(body); + let provider = DownloadBinaryProvider { + url: DownloadUrl::Fixed(server.url()), + sha256: Some(DownloadChecksum::Fixed(sha256_hex(body))), + cache_dir: Some(temp.path().join("cache")), + processor: None, + } + .with_processor_fn("empty-v1", |_artifact, _output| Ok(())); + + let error = provider + .resolve() + .expect_err("missing processed output is rejected"); + + assert!(matches!( + error, + BinaryProviderError::MissingProcessedOutput { .. } + )); +} + fn write_file(path: &Path, contents: &[u8]) { fs::write(path, contents).expect("write file"); } diff --git a/testing-framework/deployers/local/src/binary/types.rs b/testing-framework/deployers/local/src/binary/types.rs index 50c41c5..5fcd591 100644 --- a/testing-framework/deployers/local/src/binary/types.rs +++ b/testing-framework/deployers/local/src/binary/types.rs @@ -1,4 +1,8 @@ -use std::{env, fmt, iter, path::PathBuf, sync::Arc}; +use std::{ + env, fmt, iter, + path::{Path, PathBuf}, + sync::Arc, +}; use thiserror::Error; @@ -132,7 +136,64 @@ impl BuildBinaryProvider { } } -#[derive(Clone, Debug)] +/// Error returned by integration-specific download processors. +pub type DownloadProcessorError = Box; + +/// Post-download preparation step for artifacts that are not directly +/// executable. +/// +/// Implementations receive the checksum-verified downloaded artifact and must +/// materialize the executable at `output`. The stable cache key ensures that a +/// change in preparation logic invalidates the previously prepared binary. +pub trait DownloadProcessor: Send + Sync { + fn process(&self, artifact: &Path, output: &Path) -> Result<(), DownloadProcessorError>; + + fn cache_key(&self) -> &str; +} + +type DownloadProcessorCallback = + dyn Fn(&Path, &Path) -> Result<(), DownloadProcessorError> + Send + Sync; + +/// Named callback adapter for lightweight, integration-specific processing. +#[derive(Clone)] +pub struct DownloadProcessorFn { + cache_key: String, + callback: Arc, +} + +impl DownloadProcessorFn { + #[must_use] + pub fn new( + cache_key: impl Into, + callback: impl Fn(&Path, &Path) -> Result<(), DownloadProcessorError> + Send + Sync + 'static, + ) -> Self { + Self { + cache_key: cache_key.into(), + callback: Arc::new(callback), + } + } +} + +impl DownloadProcessor for DownloadProcessorFn { + fn process(&self, artifact: &Path, output: &Path) -> Result<(), DownloadProcessorError> { + (self.callback)(artifact, output) + } + + fn cache_key(&self) -> &str { + &self.cache_key + } +} + +impl fmt::Debug for DownloadProcessorFn { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("DownloadProcessorFn") + .field("cache_key", &self.cache_key) + .finish_non_exhaustive() + } +} + +#[derive(Clone)] pub struct DownloadBinaryProvider { /// Download source used to fetch the executable. pub url: DownloadUrl, @@ -143,6 +204,8 @@ pub struct DownloadBinaryProvider { /// If unset, downloads are cached under `target/.tf-binaries` in the /// current process directory. pub cache_dir: Option, + /// Optional preparation step for archives or other packaged artifacts. + pub processor: Option>, } impl DownloadBinaryProvider { @@ -152,8 +215,42 @@ impl DownloadBinaryProvider { url: DownloadUrl::Fixed(url.into()), sha256: None, cache_dir: None, + processor: None, } } + + #[must_use] + pub fn with_processor(mut self, processor: impl DownloadProcessor + 'static) -> Self { + self.processor = Some(Arc::new(processor)); + self + } + + #[must_use] + pub fn with_processor_fn( + self, + cache_key: impl Into, + callback: impl Fn(&Path, &Path) -> Result<(), DownloadProcessorError> + Send + Sync + 'static, + ) -> Self { + self.with_processor(DownloadProcessorFn::new(cache_key, callback)) + } +} + +impl fmt::Debug for DownloadBinaryProvider { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("DownloadBinaryProvider") + .field("url", &self.url) + .field("sha256", &self.sha256) + .field("cache_dir", &self.cache_dir) + .field( + "processor", + &self + .processor + .as_ref() + .map(|processor| processor.cache_key()), + ) + .finish() + } } #[derive(Clone, Debug)] @@ -253,6 +350,23 @@ pub enum BinaryProviderError { /// Actual lowercase SHA-256 checksum of the downloaded bytes. actual: String, }, + /// A download processor completed without creating its configured output. + #[error("download processor {processor} did not produce binary output {path:?}")] + MissingProcessedOutput { + /// Stable identifier of the processor that failed to produce output. + processor: String, + /// Expected temporary executable output path. + path: PathBuf, + }, + /// A configured download processor failed while preparing the executable. + #[error("download processor {processor} failed: {source}")] + DownloadProcessing { + /// Stable identifier of the processor that failed. + processor: String, + #[source] + /// Processor-specific error. + source: DownloadProcessorError, + }, /// Filesystem operation failed while preparing or resolving a binary. #[error("failed to prepare binary path {path:?}: {source}")] Io { diff --git a/testing-framework/deployers/local/src/deployer/orchestrator.rs b/testing-framework/deployers/local/src/deployer/orchestrator.rs index 18d7d23..47884c5 100644 --- a/testing-framework/deployers/local/src/deployer/orchestrator.rs +++ b/testing-framework/deployers/local/src/deployer/orchestrator.rs @@ -27,7 +27,7 @@ use tokio_retry::{ use tracing::{debug, info, warn}; use crate::{ - env::{LocalDeployerEnv, Node, wait_local_http_readiness}, + env::{LocalDeployerEnv, Node, wait_local_readiness}, external::build_external_client, keep_tempdir_from_env, manual::ManualCluster, @@ -440,7 +440,7 @@ async fn run_readiness_for_attempt( return Ok(nodes); } - match wait_local_http_readiness::(&nodes, execution.readiness_requirement).await { + match wait_local_readiness::(&nodes, execution.readiness_requirement).await { Ok(()) => { info!(attempt, "local nodes are ready"); Ok(nodes) @@ -519,7 +519,7 @@ async fn run_context_for( runtime_cleanup: Option>, node_control: Option>>, ) -> Result, ProcessDeployerError> { - if node_clients.is_empty() { + if node_clients.is_empty() && runtime_extensions.is_empty() { return Err(ProcessDeployerError::RuntimePreflight); } diff --git a/testing-framework/deployers/local/src/env/helpers.rs b/testing-framework/deployers/local/src/env/helpers.rs index 010e0b0..d639b15 100644 --- a/testing-framework/deployers/local/src/env/helpers.rs +++ b/testing-framework/deployers/local/src/env/helpers.rs @@ -10,7 +10,7 @@ use testing_framework_core::{ }; use crate::{ - binary::{BinaryProvider, BinaryProviderRef, EnvBinaryProvider}, + binary::{BinaryProvider, BinaryProviderRef, EnvBinaryProvider, PathBinaryProvider}, env::LocalBuildContext, process::{LaunchSpec, NodeEndpointPort, NodeEndpoints, ProcessSpawnError}, }; @@ -63,8 +63,8 @@ impl LocalNodePorts { } } -#[derive(Clone, Debug)] /// Peer node view used while constructing local configs. +#[derive(Clone, Debug)] pub struct LocalPeerNode { index: usize, network_port: u16, @@ -96,6 +96,16 @@ impl LocalPeerNode { } } +/// How a rendered local config file path is passed to the child process. +#[derive(Clone, Default)] +pub enum LocalConfigArgMode { + /// Pass the config file as a flag pair, for example `--config config.yaml`. + #[default] + Flag, + /// Pass the config file path as a positional argument. + Positional, +} + /// Standard local process description for one node binary plus one config file. #[derive(Clone)] pub struct LocalProcessSpec { @@ -105,6 +115,8 @@ pub struct LocalProcessSpec { pub config_file_name: String, /// CLI flag used to point the process at `config_file_name`. pub config_arg: String, + /// Controls whether `config_arg` is emitted or the file is positional. + pub config_arg_mode: LocalConfigArgMode, /// Extra CLI arguments passed after the config flag. pub extra_args: Vec, /// Extra environment variables for the child process. @@ -119,16 +131,32 @@ impl LocalProcessSpec { binary: Arc::new(EnvBinaryProvider::new(binary_env_var)), config_file_name: "config.yaml".to_owned(), config_arg: "--config".to_owned(), + config_arg_mode: LocalConfigArgMode::Flag, extra_args: Vec::new(), env: Vec::new(), } } + /// Sets an explicit binary path for this launch. + #[must_use] + pub fn with_binary_path(self, path: impl Into) -> Self { + self.with_binary_provider(PathBinaryProvider::new(path)) + } + /// Overrides the config file name and CLI flag used to pass it. #[must_use] pub fn with_config_file(mut self, file_name: &str, arg: &str) -> Self { self.config_file_name = file_name.to_owned(); self.config_arg = arg.to_owned(); + self.config_arg_mode = LocalConfigArgMode::Flag; + self + } + + /// Overrides the config file name and passes it as a positional argument. + #[must_use] + pub fn with_positional_config_file(mut self, file_name: &str) -> Self { + self.config_file_name = file_name.to_owned(); + self.config_arg_mode = LocalConfigArgMode::Positional; self } @@ -404,7 +432,7 @@ pub(crate) fn rendered_config_launch_spec( spec: &LocalProcessSpec, ) -> Result { let binary = spec.binary.resolve()?; - let mut args = vec![spec.config_arg.clone(), spec.config_file_name.clone()]; + let mut args = config_file_args(spec); args.extend(spec.extra_args.iter().cloned()); Ok(LaunchSpec { @@ -417,3 +445,39 @@ pub(crate) fn rendered_config_launch_spec( env: spec.env.clone(), }) } +fn config_file_args(spec: &LocalProcessSpec) -> Vec { + match spec.config_arg_mode { + LocalConfigArgMode::Flag => vec![spec.config_arg.clone(), spec.config_file_name.clone()], + LocalConfigArgMode::Positional => vec![spec.config_file_name.clone()], + } +} + +#[cfg(test)] +mod tests { + use super::{LocalProcessSpec, text_config_launch_spec}; + + #[test] + fn launch_spec_uses_flag_config_by_default() { + let temp = tempfile::tempdir().expect("temp dir"); + let binary = temp.path().join("app"); + std::fs::write(&binary, b"binary").expect("test binary"); + let spec = LocalProcessSpec::new("APP_BIN").with_binary_path(binary); + let launch = text_config_launch_spec("config", &spec).expect("launch spec"); + + assert_eq!(launch.args, ["--config", "config.yaml"]); + } + + #[test] + fn launch_spec_can_use_positional_config_path() { + let temp = tempfile::tempdir().expect("temp dir"); + let binary = temp.path().join("app"); + std::fs::write(&binary, b"binary").expect("test binary"); + let spec = LocalProcessSpec::new("APP_BIN") + .with_binary_path(binary) + .with_positional_config_file("app.json") + .with_args(["--port".to_owned(), "8080".to_owned()]); + let launch = text_config_launch_spec("config", &spec).expect("launch spec"); + + assert_eq!(launch.args, ["app.json", "--port", "8080"]); + } +} diff --git a/testing-framework/deployers/local/src/env/mod.rs b/testing-framework/deployers/local/src/env/mod.rs index 9f52e5f..34993cd 100644 --- a/testing-framework/deployers/local/src/env/mod.rs +++ b/testing-framework/deployers/local/src/env/mod.rs @@ -1,13 +1,14 @@ use std::{ collections::HashMap, - net::{Ipv4Addr, SocketAddr}, + net::{Ipv4Addr, SocketAddr, TcpStream}, path::{Path, PathBuf}, + time::{Duration, Instant}, }; use async_trait::async_trait; use testing_framework_core::scenario::{ Application, DynError, HttpReadinessRequirement, ReadinessError, StartNodeOptions, - wait_for_http_ports_with_requirement, + wait_for_http_ports_with_requirement_and_timeout, }; use crate::{ @@ -20,14 +21,27 @@ mod helpers; mod tests; pub use helpers::{ - BuiltNodeConfig, LocalNodePorts, LocalPeerNode, LocalProcessSpec, NodeConfigEntry, - build_indexed_http_peers, build_indexed_node_configs, build_launch_spec_with_args, - build_local_cluster_node_config, build_local_peer_nodes, default_yaml_launch_spec, - discovered_node_access, preallocate_ports, reserve_local_node_ports, + BuiltNodeConfig, LocalConfigArgMode, LocalNodePorts, LocalPeerNode, LocalProcessSpec, + NodeConfigEntry, build_indexed_http_peers, build_indexed_node_configs, + build_launch_spec_with_args, build_local_cluster_node_config, build_local_peer_nodes, + default_yaml_launch_spec, discovered_node_access, preallocate_ports, reserve_local_node_ports, single_http_node_endpoints, text_config_launch_spec, text_node_config, yaml_config_launch_spec, yaml_node_config, }; +const TCP_READINESS_TIMEOUT: Duration = Duration::from_secs(60); +const TCP_READINESS_POLL_INTERVAL: Duration = Duration::from_millis(200); +const TCP_CONNECT_TIMEOUT: Duration = Duration::from_millis(100); + +/// Readiness probe shape used by the local process deployer. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum LocalReadinessProbe { + /// Probe `http://127.0.0.1:/` with GET. + HttpGet { path: &'static str }, + /// Probe that the API port accepts TCP connections. + Tcp, +} + /// Context passed while building a local node config. pub struct LocalBuildContext<'a, E: Application> { /// Full deployment topology for the current scenario. @@ -323,6 +337,13 @@ where ::node_readiness_path() } + /// Returns the readiness probe used for local process startup. + fn readiness_probe() -> LocalReadinessProbe { + LocalReadinessProbe::HttpGet { + path: Self::readiness_endpoint_path(), + } + } + /// Waits for any additional cluster-specific stabilization after the HTTP /// readiness probe succeeds. async fn wait_readiness_stable(_nodes: &[Node]) -> Result<(), DynError> { @@ -376,6 +397,13 @@ where ::node_readiness_path() } + /// Returns the readiness probe used for local process startup. + fn readiness_probe() -> LocalReadinessProbe { + LocalReadinessProbe::HttpGet { + path: Self::readiness_endpoint_path(), + } + } + /// Waits for any additional cluster-specific stabilization after the HTTP /// readiness probe succeeds. async fn wait_readiness_stable(_nodes: &[Node]) -> Result<(), DynError> { @@ -435,6 +463,10 @@ where T::readiness_endpoint_path() } + fn readiness_probe() -> LocalReadinessProbe { + T::readiness_probe() + } + async fn wait_readiness_stable(nodes: &[Node]) -> Result<(), DynError> { T::wait_readiness_stable(nodes).await } @@ -490,13 +522,9 @@ pub(crate) fn node_peer_port(node: &Node) -> u16 { E::node_peer_port(node) } -pub(crate) fn readiness_endpoint_path() -> &'static str { - E::readiness_endpoint_path() -} - -/// Waits for local HTTP readiness across the provided nodes and then applies +/// Waits for local readiness across the provided nodes and then applies /// any app-specific stabilization hook. -pub async fn wait_local_http_readiness( +pub async fn wait_local_readiness( nodes: &[Node], requirement: HttpReadinessRequirement, ) -> Result<(), ReadinessError> { @@ -505,13 +533,119 @@ pub async fn wait_local_http_readiness( .map(|node| node.endpoints().api.port()) .collect(); - wait_for_http_ports_with_requirement(&ports, E::readiness_endpoint_path(), requirement).await?; + wait_for_local_readiness_ports::(&ports, requirement, None).await?; E::wait_readiness_stable(nodes) .await .map_err(|source| ReadinessError::ClusterStable { source }) } +/// Backward-compatible name for local readiness checks that used to be +/// HTTP-only. +pub async fn wait_local_http_readiness( + nodes: &[Node], + requirement: HttpReadinessRequirement, +) -> Result<(), ReadinessError> { + wait_local_readiness::(nodes, requirement).await +} + +pub(crate) async fn wait_for_local_readiness_ports( + ports: &[u16], + requirement: HttpReadinessRequirement, + timeout: Option, +) -> Result<(), ReadinessError> { + match E::readiness_probe() { + LocalReadinessProbe::HttpGet { path } => { + wait_for_http_ports_with_requirement_and_timeout(ports, path, requirement, timeout) + .await + } + LocalReadinessProbe::Tcp => { + wait_for_tcp_ports_with_requirement(ports, requirement, timeout).await + } + } +} + +async fn wait_for_tcp_ports_with_requirement( + ports: &[u16], + requirement: HttpReadinessRequirement, + timeout: Option, +) -> Result<(), ReadinessError> { + if ports.is_empty() { + return Ok(()); + } + + let timeout = timeout.unwrap_or(TCP_READINESS_TIMEOUT); + let deadline = Instant::now() + timeout; + + loop { + let statuses = collect_tcp_statuses(ports); + if tcp_requirement_satisfied(&statuses, requirement) { + return Ok(()); + } + + if Instant::now() >= deadline { + return Err(ReadinessError::ProbeTimeout { + message: format_tcp_timeout_message(&statuses, requirement), + }); + } + + tokio::time::sleep(TCP_READINESS_POLL_INTERVAL).await; + } +} + +fn collect_tcp_statuses(ports: &[u16]) -> Vec<(u16, bool)> { + ports + .iter() + .map(|port| (*port, tcp_port_ready(*port))) + .collect() +} + +fn tcp_port_ready(port: u16) -> bool { + let addr = SocketAddr::from((Ipv4Addr::LOCALHOST, port)); + TcpStream::connect_timeout(&addr, TCP_CONNECT_TIMEOUT).is_ok() +} + +fn tcp_requirement_satisfied( + statuses: &[(u16, bool)], + requirement: HttpReadinessRequirement, +) -> bool { + let ready = tcp_ready_count(statuses); + + match requirement { + HttpReadinessRequirement::AllNodesReady => ready == statuses.len(), + HttpReadinessRequirement::AnyNodeReady => ready >= 1, + HttpReadinessRequirement::AtLeast(min_ready) => ready >= min_ready, + } +} + +fn tcp_ready_count(statuses: &[(u16, bool)]) -> usize { + statuses.iter().filter(|(_, ready)| *ready).count() +} + +fn format_tcp_timeout_message( + statuses: &[(u16, bool)], + requirement: HttpReadinessRequirement, +) -> String { + let ready = tcp_ready_count(statuses); + let failing = statuses + .iter() + .filter_map(|(port, ok)| (!ok).then_some(port.to_string())) + .collect::>() + .join(", "); + + format!( + "timed out waiting for TCP readiness {:?}; ready={}, total={}, failing ports: {}", + requirement, + ready, + statuses.len(), + if failing.is_empty() { + "" + } else { + &failing + } + ) +} + /// Spawns a local process node from an already prepared config value. pub async fn spawn_node_from_config( label: String, diff --git a/testing-framework/deployers/local/src/env/tests.rs b/testing-framework/deployers/local/src/env/tests.rs index 756a9d1..044adc1 100644 --- a/testing-framework/deployers/local/src/env/tests.rs +++ b/testing-framework/deployers/local/src/env/tests.rs @@ -1,4 +1,8 @@ -use std::sync::atomic::{AtomicUsize, Ordering}; +use std::{ + net::TcpListener, + sync::atomic::{AtomicUsize, Ordering}, + time::Duration, +}; use testing_framework_core::{ scenario::{Application, DynError, HttpReadinessRequirement}, @@ -22,6 +26,7 @@ impl DeploymentDescriptor for DummyTopology { } struct DummyEnv; +struct TcpEnv; #[async_trait::async_trait] impl Application for DummyEnv { @@ -30,6 +35,13 @@ impl Application for DummyEnv { type NodeConfig = DummyConfig; } +#[async_trait::async_trait] +impl Application for TcpEnv { + type Deployment = DummyTopology; + type NodeClient = (); + type NodeConfig = DummyConfig; +} + #[async_trait::async_trait] impl LocalDeployerEnv for DummyEnv { fn build_node_config( @@ -69,6 +81,13 @@ impl LocalDeployerEnv for DummyEnv { } } +#[async_trait::async_trait] +impl LocalDeployerEnv for TcpEnv { + fn readiness_probe() -> LocalReadinessProbe { + LocalReadinessProbe::Tcp + } +} + fn build_dummy_node() -> Result, DynError> { unreachable!("not used in this test") } @@ -99,8 +118,22 @@ async fn dummy_wait_stable() -> Result<(), DynError> { async fn empty_cluster_still_runs_stability_hook() { STABLE_CALLS.store(0, Ordering::SeqCst); let nodes: Vec> = Vec::new(); - wait_local_http_readiness::(&nodes, HttpReadinessRequirement::AllNodesReady) + wait_local_readiness::(&nodes, HttpReadinessRequirement::AllNodesReady) .await .expect("empty cluster should be considered ready"); assert_eq!(STABLE_CALLS.load(Ordering::SeqCst), 1); } + +#[tokio::test] +async fn tcp_readiness_probe_accepts_bound_port() { + let listener = TcpListener::bind("127.0.0.1:0").expect("bind tcp listener"); + let port = listener.local_addr().expect("listener addr").port(); + + wait_for_local_readiness_ports::( + &[port], + HttpReadinessRequirement::AllNodesReady, + Some(Duration::from_secs(1)), + ) + .await + .expect("bound TCP port should be ready"); +} diff --git a/testing-framework/deployers/local/src/lib.rs b/testing-framework/deployers/local/src/lib.rs index 14eda23..72692b8 100644 --- a/testing-framework/deployers/local/src/lib.rs +++ b/testing-framework/deployers/local/src/lib.rs @@ -8,17 +8,18 @@ pub mod process; pub use binary::{ BinaryProvider, BinaryProviderError, BinaryProviderRef, BuildBinaryProvider, BuildCommand, - DownloadBinaryProvider, DownloadChecksum, DownloadUrl, EnvBinaryProvider, - FallbackBinaryProvider, PathBinaryProvider, + DownloadBinaryProvider, DownloadChecksum, DownloadProcessor, DownloadProcessorError, + DownloadProcessorFn, DownloadUrl, EnvBinaryProvider, FallbackBinaryProvider, + PathBinaryProvider, }; pub use deployer::{ProcessDeployer, ProcessDeployerError}; pub use env::{ - BuiltNodeConfig, LocalBinaryApp, LocalBuildContext, LocalDeployerEnv, LocalNodePorts, - LocalPeerNode, LocalProcessSpec, NodeConfigEntry, build_indexed_http_peers, - build_indexed_node_configs, build_local_cluster_node_config, build_local_peer_nodes, - default_yaml_launch_spec, discovered_node_access, preallocate_ports, reserve_local_node_ports, - single_http_node_endpoints, text_config_launch_spec, text_node_config, yaml_config_launch_spec, - yaml_node_config, + BuiltNodeConfig, LocalBinaryApp, LocalBuildContext, LocalConfigArgMode, LocalDeployerEnv, + LocalNodePorts, LocalPeerNode, LocalProcessSpec, LocalReadinessProbe, NodeConfigEntry, + build_indexed_http_peers, build_indexed_node_configs, build_local_cluster_node_config, + build_local_peer_nodes, default_yaml_launch_spec, discovered_node_access, preallocate_ports, + reserve_local_node_ports, single_http_node_endpoints, text_config_launch_spec, + text_node_config, yaml_config_launch_spec, yaml_node_config, }; pub use manual::{ManualCluster, ManualClusterError}; pub use node_control::{NodeManager, NodeManagerError, NodeManagerSeed}; diff --git a/testing-framework/deployers/local/src/node_control/mod.rs b/testing-framework/deployers/local/src/node_control/mod.rs index 24a9a73..8dccf56 100644 --- a/testing-framework/deployers/local/src/node_control/mod.rs +++ b/testing-framework/deployers/local/src/node_control/mod.rs @@ -4,8 +4,8 @@ use std::{ }; use testing_framework_core::scenario::{ - Application, DynError, NodeClients, NodeControlHandle, NodeRuntimeOptions, ReadinessError, - StartNodeOptions, StartedNode, wait_for_http_ports, wait_for_http_ports_with_timeout, + Application, DynError, HttpReadinessRequirement, NodeClients, NodeControlHandle, + NodeRuntimeOptions, ReadinessError, StartNodeOptions, StartedNode, }; use thiserror::Error; @@ -13,7 +13,7 @@ use crate::{ env::{ LocalDeployerEnv, Node, build_initial_node_configs, build_launch_spec_with_args, build_node_from_template, initial_persist_dir, initial_snapshot_dir, node_peer_port, - readiness_endpoint_path, spawn_node_from_config, + spawn_node_from_config, wait_for_local_readiness_ports, }, process::ProcessSpawnError, }; @@ -214,15 +214,16 @@ impl NodeManager { return Ok(()); } - wait_for_http_ports(&ports, readiness_endpoint_path::()).await + wait_for_local_readiness_ports::(&ports, HttpReadinessRequirement::AllNodesReady, None) + .await } pub async fn wait_node_ready(&self, name: &str) -> Result<(), NodeManagerError> { let target = self.readiness_target(name)?; - wait_for_http_ports_with_timeout( + wait_for_local_readiness_ports::( &[target.port], - readiness_endpoint_path::(), + HttpReadinessRequirement::AllNodesReady, target.runtime.start_timeout, ) .await