From 6f13873cb7627434671ac9026c571243a7b78027 Mon Sep 17 00:00:00 2001 From: andrussal Date: Fri, 17 Jul 2026 06:00:47 +0200 Subject: [PATCH] feat(tf): add composable app deployments --- Cargo.lock | 19 +- Cargo.toml | 2 + testing-framework/app/Cargo.toml | 21 + testing-framework/app/src/cleanup.rs | 81 ++++ testing-framework/app/src/context.rs | 323 +++++++++++++++ testing-framework/app/src/deployment.rs | 30 ++ testing-framework/app/src/error.rs | 29 ++ testing-framework/app/src/extension.rs | 233 +++++++++++ testing-framework/app/src/host.rs | 55 +++ testing-framework/app/src/lib.rs | 35 ++ testing-framework/app/src/local.rs | 238 +++++++++++ testing-framework/app/src/process.rs | 387 ++++++++++++++++++ testing-framework/app/src/registry.rs | 281 +++++++++++++ .../core/src/scenario/runtime/extensions.rs | 5 + .../local/src/deployer/orchestrator.rs | 6 +- .../deployers/local/src/env/helpers.rs | 70 +++- .../deployers/local/src/env/mod.rs | 160 +++++++- .../deployers/local/src/env/tests.rs | 37 +- testing-framework/deployers/local/src/lib.rs | 12 +- .../deployers/local/src/node_control/mod.rs | 13 +- 20 files changed, 2000 insertions(+), 37 deletions(-) create mode 100644 testing-framework/app/Cargo.toml create mode 100644 testing-framework/app/src/cleanup.rs create mode 100644 testing-framework/app/src/context.rs create mode 100644 testing-framework/app/src/deployment.rs create mode 100644 testing-framework/app/src/error.rs create mode 100644 testing-framework/app/src/extension.rs create mode 100644 testing-framework/app/src/host.rs create mode 100644 testing-framework/app/src/lib.rs create mode 100644 testing-framework/app/src/local.rs create mode 100644 testing-framework/app/src/process.rs create mode 100644 testing-framework/app/src/registry.rs diff --git a/Cargo.lock b/Cargo.lock index 385591d..2f8c5eb 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", ] @@ -3602,6 +3602,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..6f56b42 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -30,6 +30,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 +59,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/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/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 fd4751a..72692b8 100644 --- a/testing-framework/deployers/local/src/lib.rs +++ b/testing-framework/deployers/local/src/lib.rs @@ -14,12 +14,12 @@ pub use binary::{ }; 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