From e5dc6a8d4678f9ea236c7cc9ed424093ab1b0526 Mon Sep 17 00:00:00 2001 From: andrussal Date: Sat, 18 Jul 2026 12:04:25 +0200 Subject: [PATCH] feat(examples): add multi-app test stack --- Cargo.lock | 34 +- Cargo.toml | 4 +- examples/README.md | 4 +- examples/kvstore/kvstore-node/src/client.rs | 5 + examples/multi_app/README.md | 52 ++- examples/multi_app/e2e/Cargo.toml | 11 + examples/multi_app/e2e/src/lib.rs | 1 + .../multi_app/e2e/tests/local_happy_path.rs | 23 ++ examples/multi_app/examples/Cargo.toml | 22 -- .../examples/src/bin/typed_stack_smoke.rs | 269 -------------- examples/multi_app/fixture/Cargo.toml | 18 + examples/multi_app/fixture/src/lib.rs | 348 ++++++++++++++++++ examples/multi_app/job-worker/Cargo.toml | 16 + examples/multi_app/job-worker/src/main.rs | 125 +++++++ examples/queue/queue-node/src/client.rs | 5 + examples/queue/testing/integration/Cargo.toml | 1 + examples/queue/testing/integration/src/app.rs | 29 ++ .../testing/integration/src/local_env.rs | 37 +- 18 files changed, 685 insertions(+), 319 deletions(-) create mode 100644 examples/multi_app/e2e/Cargo.toml create mode 100644 examples/multi_app/e2e/src/lib.rs create mode 100644 examples/multi_app/e2e/tests/local_happy_path.rs delete mode 100644 examples/multi_app/examples/Cargo.toml delete mode 100644 examples/multi_app/examples/src/bin/typed_stack_smoke.rs create mode 100644 examples/multi_app/fixture/Cargo.toml create mode 100644 examples/multi_app/fixture/src/lib.rs create mode 100644 examples/multi_app/job-worker/Cargo.toml create mode 100644 examples/multi_app/job-worker/src/main.rs diff --git a/Cargo.lock b/Cargo.lock index 9dbccbd..c9d44c3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1925,17 +1925,42 @@ dependencies = [ ] [[package]] -name = "multi-app-examples" +name = "multi-app-e2e" +version = "0.1.0" +dependencies = [ + "multi-app-fixture", + "testing-framework-app", + "testing-framework-core", + "tokio", +] + +[[package]] +name = "multi-app-fixture" version = "0.1.0" dependencies = [ - "anyhow", "async-trait", + "kvstore-node", "kvstore-runtime-ext", - "openraft-kv-runtime-ext", - "openraft-kv-runtime-workloads", + "queue-runtime-ext", + "reqwest", "serde", "testing-framework-app", "testing-framework-core", + "testing-framework-runner-local", + "tokio", + "tracing", +] + +[[package]] +name = "multi-app-job-worker" +version = "0.1.0" +dependencies = [ + "anyhow", + "axum", + "kvstore-node", + "queue-node", + "reqwest", + "serde", "tokio", "tracing", "tracing-subscriber", @@ -2701,6 +2726,7 @@ dependencies = [ "async-trait", "queue-node", "serde", + "testing-framework-app", "testing-framework-core", "testing-framework-runner-compose", "testing-framework-runner-local", diff --git a/Cargo.toml b/Cargo.toml index 05d8355..44b89a2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -12,7 +12,9 @@ members = [ "examples/metrics_counter/metrics-counter-node", "examples/metrics_counter/testing/integration", "examples/metrics_counter/testing/workloads", - "examples/multi_app/examples", + "examples/multi_app/e2e", + "examples/multi_app/fixture", + "examples/multi_app/job-worker", "examples/nats/examples", "examples/nats/testing/integration", "examples/nats/testing/workloads", diff --git a/examples/README.md b/examples/README.md index c751212..49762c3 100644 --- a/examples/README.md +++ b/examples/README.md @@ -9,8 +9,8 @@ Canonical app-layer examples: `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 +- `multi-app-e2e`: composed queue, worker, and result-store stack exercised + through integration tests The older direct `ScenarioBuilder` examples are still useful for backend-specific coverage: diff --git a/examples/kvstore/kvstore-node/src/client.rs b/examples/kvstore/kvstore-node/src/client.rs index 8276795..1bae5f3 100644 --- a/examples/kvstore/kvstore-node/src/client.rs +++ b/examples/kvstore/kvstore-node/src/client.rs @@ -16,6 +16,11 @@ impl KvHttpClient { } } + #[must_use] + pub const fn base_url(&self) -> &Url { + &self.base_url + } + pub async fn get(&self, path: &str) -> anyhow::Result { let url = self.base_url.join(path)?; let response = self.client.get(url).send().await?.error_for_status()?; diff --git a/examples/multi_app/README.md b/examples/multi_app/README.md index abec9bb..9de1e86 100644 --- a/examples/multi_app/README.md +++ b/examples/multi_app/README.md @@ -1,39 +1,49 @@ -# Multi-App Examples +# Multi-App Acceptance Tests -This directory shows the canonical app-layer pattern for composed systems. +This directory provides a reusable fixture and end-to-end coverage for composed +systems. The fixture deploys a two-node queue, a two-node result store, and a +worker process that consumes queued jobs and records their completion: -Use this shape when a scenario needs several apps or resources that should feel -like one system to the workload: +```text +workload -> queue -> worker -> result store -> expectation +``` + +The local happy-path test drives the fixture through a scenario: ```rust let mut scenario = AppHost::scenario() - .with_app(ExampleStackApp::new()) - .with_workload(ExampleStackWorkload::new()) + .with_app(JobStackApp::new()) + .with_workload(EnqueueJobs::new(10)) + .with_expectation(AllJobsCompleted::new(10)) .build()?; ``` -The stack app deploys child apps, exposes their typed handles, and returns a -composed stack handle: +`multi-app-fixture` owns the stack definition. It 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()); +let queue = ctx.deploy_and_expose(self.queue).await?; +let results = ctx.deploy_and_expose(self.results).await?; +let worker = ctx + .deploy_and_expose(JobWorkerApp::new(queue_url, results_url)) + .await?; -ctx.expose(store)?; -ctx.expose(consensus)?; -ctx.expose(wallet)?; +let stack = JobStackHandle { queue, results, worker }; ctx.expose(stack.clone())?; ``` -Workloads then request concrete handles: +Workloads, expectations, and later lifecycle tests request the composed handle: ```rust -let stack = ctx.require_app::()?; -let wallet = ctx.require_app::()?; +let stack = ctx.require_app::()?; ``` +The worker is part of the deployed system, rather than test code moving data +between otherwise unrelated applications. It lives in the +`multi-app-job-worker` crate, receives the queue and result-store endpoints from +the parent deployment, and is started through `LocalProcessApp`. Reverse +cleanup stops the worker before either dependency. + Resource lifecycle comes from the TF adapter used by a child deployment: - `LocalProcessApp` manages one local binary process. @@ -54,3 +64,9 @@ 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. + +Run the local end-to-end test from the workspace root: + +```shell +cargo test -p multi-app-e2e --test local_happy_path +``` diff --git a/examples/multi_app/e2e/Cargo.toml b/examples/multi_app/e2e/Cargo.toml new file mode 100644 index 0000000..8173d91 --- /dev/null +++ b/examples/multi_app/e2e/Cargo.toml @@ -0,0 +1,11 @@ +[package] +edition.workspace = true +license.workspace = true +name = "multi-app-e2e" +version.workspace = true + +[dependencies] +multi-app-fixture = { path = "../fixture" } +testing-framework-app = { workspace = true } +testing-framework-core = { workspace = true } +tokio = { workspace = true, features = ["full"] } diff --git a/examples/multi_app/e2e/src/lib.rs b/examples/multi_app/e2e/src/lib.rs new file mode 100644 index 0000000..559e65d --- /dev/null +++ b/examples/multi_app/e2e/src/lib.rs @@ -0,0 +1 @@ +//! End-to-end coverage for composed testing-framework applications. diff --git a/examples/multi_app/e2e/tests/local_happy_path.rs b/examples/multi_app/e2e/tests/local_happy_path.rs new file mode 100644 index 0000000..2bb54b5 --- /dev/null +++ b/examples/multi_app/e2e/tests/local_happy_path.rs @@ -0,0 +1,23 @@ +use std::time::Duration; + +use multi_app_fixture::{AllJobsCompleted, EnqueueJobs, JobStackApp}; +use testing_framework_app::{AppHost, AppHostLocalDeployer, AppScenarioBuilderExt}; +use testing_framework_core::scenario::{Deployer, DynError}; + +const JOB_COUNT: usize = 10; + +#[tokio::test] +async fn processes_queued_jobs_and_converges_results() -> Result<(), DynError> { + let mut scenario = AppHost::scenario() + .with_app(JobStackApp::new()) + .with_run_duration(Duration::from_secs(10)) + .with_workload(EnqueueJobs::new(JOB_COUNT)) + .with_expectation(AllJobsCompleted::new(JOB_COUNT)) + .build()?; + + let deployer = AppHostLocalDeployer::default(); + let runner = deployer.deploy(&scenario).await?; + runner.run(&mut scenario).await?; + + Ok(()) +} diff --git a/examples/multi_app/examples/Cargo.toml b/examples/multi_app/examples/Cargo.toml deleted file mode 100644 index bd40515..0000000 --- a/examples/multi_app/examples/Cargo.toml +++ /dev/null @@ -1,22 +0,0 @@ -[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 deleted file mode 100644 index a857b0b..0000000 --- a/examples/multi_app/examples/src/bin/typed_stack_smoke.rs +++ /dev/null @@ -1,269 +0,0 @@ -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/multi_app/fixture/Cargo.toml b/examples/multi_app/fixture/Cargo.toml new file mode 100644 index 0000000..bec0f80 --- /dev/null +++ b/examples/multi_app/fixture/Cargo.toml @@ -0,0 +1,18 @@ +[package] +edition.workspace = true +license.workspace = true +name = "multi-app-fixture" +version.workspace = true + +[dependencies] +async-trait = { workspace = true } +kvstore-node = { path = "../../kvstore/kvstore-node" } +kvstore-runtime-ext = { path = "../../kvstore/testing/integration" } +queue-runtime-ext = { path = "../../queue/testing/integration" } +reqwest = { workspace = true, features = ["json"] } +serde = { workspace = true } +testing-framework-app = { workspace = true } +testing-framework-core = { workspace = true } +testing-framework-runner-local = { workspace = true } +tokio = { workspace = true, features = ["full"] } +tracing = { workspace = true } diff --git a/examples/multi_app/fixture/src/lib.rs b/examples/multi_app/fixture/src/lib.rs new file mode 100644 index 0000000..f07e56e --- /dev/null +++ b/examples/multi_app/fixture/src/lib.rs @@ -0,0 +1,348 @@ +use std::{path::PathBuf, time::Duration}; + +use async_trait::async_trait; +use kvstore_runtime_ext::{KvEnv, KvLocalApp}; +use queue_runtime_ext::{QueueEnv, QueueLocalApp}; +use reqwest::Url; +use serde::{Deserialize, Serialize}; +use testing_framework_app::{ + AppDeployment, AppHostEnv, AppRunContextExt, DeployContext, LocalAppCluster, LocalProcessApp, + LocalProcessHandle, +}; +use testing_framework_core::scenario::{DynError, Expectation, RunContext, Workload}; +use testing_framework_runner_local::{ + BinaryProvider, BinaryProviderRef, BuildBinaryProvider, BuildCommand, EnvBinaryProvider, + FallbackBinaryProvider, LaunchSpec, NodeEndpoints, allocate_available_port, +}; +use tokio::time::Instant; +use tracing::info; + +#[derive(Clone)] +pub struct JobStackApp { + queue: QueueLocalApp, + results: KvLocalApp, +} + +impl JobStackApp { + #[must_use] + pub fn new() -> Self { + Self::with_cluster_sizes(2, 2) + } + + #[must_use] + pub fn with_cluster_sizes(queue_nodes: usize, result_nodes: usize) -> Self { + Self { + queue: QueueLocalApp::nodes(queue_nodes), + results: KvLocalApp::nodes(result_nodes), + } + } +} + +impl Default for JobStackApp { + fn default() -> Self { + Self::new() + } +} + +#[async_trait] +impl AppDeployment for JobStackApp { + type Handle = JobStackHandle; + + async fn deploy(self, ctx: &mut DeployContext) -> Result { + let queue = ctx.deploy_and_expose(self.queue).await?; + let results = ctx.deploy_and_expose(self.results).await?; + + let queue_url = queue + .first_client() + .ok_or("queue cluster has no clients")? + .base_url() + .clone(); + let results_url = results + .first_client() + .ok_or("result store has no clients")? + .base_url() + .clone(); + let worker = ctx + .deploy_and_expose(JobWorkerApp::new(queue_url, results_url)) + .await?; + + let stack = JobStackHandle { + queue, + results, + worker, + }; + ctx.expose(stack.clone())?; + + Ok(stack) + } +} + +#[derive(Clone)] +pub struct JobStackHandle { + queue: LocalAppCluster, + results: LocalAppCluster, + worker: LocalProcessHandle, +} + +impl JobStackHandle { + #[must_use] + pub const fn queue(&self) -> &LocalAppCluster { + &self.queue + } + + #[must_use] + pub const fn results(&self) -> &LocalAppCluster { + &self.results + } + + #[must_use] + pub const fn worker(&self) -> &LocalProcessHandle { + &self.worker + } +} + +struct JobWorkerApp { + queue_url: Url, + results_url: Url, +} + +impl JobWorkerApp { + const fn new(queue_url: Url, results_url: Url) -> Self { + Self { + queue_url, + results_url, + } + } +} + +#[async_trait] +impl AppDeployment for JobWorkerApp { + type Handle = LocalProcessHandle; + + async fn deploy(self, ctx: &mut DeployContext) -> Result { + let health_port = allocate_available_port()?; + let client = WorkerClient::new(health_port)?; + let launch = LaunchSpec { + binary: worker_binary_provider().resolve()?, + args: vec![ + "--queue-url".to_owned(), + self.queue_url.to_string(), + "--results-url".to_owned(), + self.results_url.to_string(), + "--health-port".to_owned(), + health_port.to_string(), + ], + ..LaunchSpec::default() + }; + let process = LocalProcessApp::new( + "job-worker", + launch, + NodeEndpoints::from_api_port(health_port), + client, + ) + .with_readiness(|_, client| async move { client.wait_ready().await }); + + ctx.deploy(process).await + } +} + +#[derive(Clone)] +pub struct WorkerClient { + health_url: Url, + client: reqwest::Client, +} + +impl WorkerClient { + fn new(port: u16) -> Result { + Ok(Self { + health_url: Url::parse(&format!("http://127.0.0.1:{port}/health/ready"))?, + client: reqwest::Client::new(), + }) + } + + async fn wait_ready(&self) -> Result<(), DynError> { + let deadline = Instant::now() + Duration::from_secs(10); + + while Instant::now() < deadline { + if self + .client + .get(self.health_url.clone()) + .send() + .await + .is_ok_and(|response| response.status().is_success()) + { + return Ok(()); + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + + Err("job worker did not become ready".into()) + } +} + +fn worker_binary_provider() -> FallbackBinaryProvider { + let workspace = workspace_root(); + let providers: [BinaryProviderRef; 2] = [ + std::sync::Arc::new(EnvBinaryProvider::new("MULTI_APP_JOB_WORKER_BIN")), + std::sync::Arc::new(BuildBinaryProvider { + command: BuildCommand::new("cargo").with_args([ + "build", + "-p", + "multi-app-job-worker", + "--bin", + "multi-app-job-worker", + ]), + output_path: PathBuf::from(format!( + "target/debug/multi-app-job-worker{}", + 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("../../..") +} + +#[derive(Clone)] +pub struct EnqueueJobs { + count: usize, +} + +impl EnqueueJobs { + #[must_use] + pub const fn new(count: usize) -> Self { + Self { count } + } +} + +#[async_trait] +impl Workload for EnqueueJobs { + fn name(&self) -> &str { + "enqueue_jobs" + } + + async fn start(&self, ctx: &RunContext) -> Result<(), DynError> { + let stack = ctx.require_app::()?; + let queue = stack + .queue + .first_client() + .ok_or("queue cluster has no clients")?; + + for index in 0..self.count { + let response: EnqueueResponse = queue + .post( + "/queue/enqueue", + &EnqueueRequest { + payload: job_key(index), + }, + ) + .await?; + if !response.accepted { + return Err(format!("queue rejected job {index}").into()); + } + } + + info!(jobs = self.count, "jobs enqueued"); + Ok(()) + } +} + +#[derive(Clone)] +pub struct AllJobsCompleted { + count: usize, + timeout: Duration, +} + +impl AllJobsCompleted { + #[must_use] + pub const fn new(count: usize) -> Self { + Self { + count, + timeout: Duration::from_secs(20), + } + } + + #[must_use] + pub const fn timeout(mut self, timeout: Duration) -> Self { + self.timeout = timeout; + self + } +} + +#[async_trait] +impl Expectation for AllJobsCompleted { + fn name(&self) -> &str { + "all_jobs_completed" + } + + async fn evaluate(&mut self, ctx: &RunContext) -> Result<(), DynError> { + let stack = ctx.require_app::()?; + let clients = stack.results.clients(); + if clients.is_empty() { + return Err("result store has no clients".into()); + } + + let deadline = Instant::now() + self.timeout; + while Instant::now() < deadline { + if all_results_are_visible(&clients, self.count).await? { + if !stack.worker.is_running().await { + return Err("job worker stopped before evaluation".into()); + } + info!(jobs = self.count, "all job results converged"); + return Ok(()); + } + tokio::time::sleep(Duration::from_millis(200)).await; + } + + Err(format!("job results did not converge within {:?}", self.timeout).into()) + } +} + +async fn all_results_are_visible( + clients: &[kvstore_node::KvHttpClient], + count: usize, +) -> Result { + for index in 0..count { + for client in clients { + let response: KvGetResponse = client.get(&format!("/kv/{}", job_key(index))).await?; + if response + .record + .as_ref() + .is_none_or(|record| record.value != "completed") + { + return Ok(false); + } + } + } + + Ok(true) +} + +fn job_key(index: usize) -> String { + format!("job-{index}") +} + +#[derive(Serialize)] +struct EnqueueRequest { + payload: String, +} + +#[derive(Deserialize)] +struct EnqueueResponse { + accepted: bool, +} + +#[derive(Deserialize)] +struct KvGetResponse { + record: Option, +} + +#[derive(Deserialize)] +struct ValueRecord { + value: String, +} diff --git a/examples/multi_app/job-worker/Cargo.toml b/examples/multi_app/job-worker/Cargo.toml new file mode 100644 index 0000000..03c6355 --- /dev/null +++ b/examples/multi_app/job-worker/Cargo.toml @@ -0,0 +1,16 @@ +[package] +edition.workspace = true +license.workspace = true +name = "multi-app-job-worker" +version.workspace = true + +[dependencies] +anyhow = "1.0" +axum = "0.7" +kvstore-node = { path = "../../kvstore/kvstore-node" } +queue-node = { path = "../../queue/queue-node" } +reqwest = { workspace = true, features = ["json"] } +serde = { workspace = true } +tokio = { workspace = true, features = ["full"] } +tracing = { workspace = true } +tracing-subscriber = { version = "0.3", features = ["env-filter"] } diff --git a/examples/multi_app/job-worker/src/main.rs b/examples/multi_app/job-worker/src/main.rs new file mode 100644 index 0000000..27da7db --- /dev/null +++ b/examples/multi_app/job-worker/src/main.rs @@ -0,0 +1,125 @@ +use std::{net::Ipv4Addr, time::Duration}; + +use axum::{Router, http::StatusCode, routing::get}; +use kvstore_node::KvHttpClient; +use queue_node::QueueHttpClient; +use reqwest::Url; +use serde::{Deserialize, Serialize}; +use tracing::{info, warn}; + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + tracing_subscriber::fmt() + .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) + .init(); + + let args = WorkerArgs::parse()?; + let queue = QueueHttpClient::new(args.queue_url); + let results = KvHttpClient::new(args.results_url); + + tokio::spawn(process_jobs(queue, results)); + + let app = Router::new().route("/health/ready", get(|| async { StatusCode::OK })); + let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, args.health_port)).await?; + info!(port = args.health_port, "job worker ready"); + axum::serve(listener, app).await?; + + Ok(()) +} + +async fn process_jobs(queue: QueueHttpClient, results: KvHttpClient) { + loop { + match dequeue(&queue).await { + Ok(Some(job)) => write_result(&results, &job).await, + Ok(None) => tokio::time::sleep(Duration::from_millis(100)).await, + Err(error) => { + warn!(%error, "failed to dequeue job"); + tokio::time::sleep(Duration::from_millis(250)).await; + } + } + } +} + +async fn dequeue(queue: &QueueHttpClient) -> anyhow::Result> { + let response: DequeueResponse = queue.post("/queue/dequeue", &EmptyRequest {}).await?; + Ok(response.message) +} + +async fn write_result(results: &KvHttpClient, job: &QueueMessage) { + loop { + let request = KvPutRequest { + value: "completed", + expected_version: None, + }; + match results + .put::<_, KvPutResponse>(&format!("/kv/{}", job.payload), &request) + .await + { + Ok(response) if response.applied => { + info!(job_id = job.id, job = %job.payload, "job completed"); + return; + } + Ok(_) => warn!(job_id = job.id, "result store rejected job result"), + Err(error) => warn!(job_id = job.id, %error, "failed to write job result"), + } + tokio::time::sleep(Duration::from_millis(250)).await; + } +} + +struct WorkerArgs { + queue_url: Url, + results_url: Url, + health_port: u16, +} + +impl WorkerArgs { + fn parse() -> anyhow::Result { + let mut queue_url = None; + let mut results_url = None; + let mut health_port = None; + let mut args = std::env::args().skip(1); + + while let Some(flag) = args.next() { + let value = args + .next() + .ok_or_else(|| anyhow::anyhow!("missing value for {flag}"))?; + match flag.as_str() { + "--queue-url" => queue_url = Some(Url::parse(&value)?), + "--results-url" => results_url = Some(Url::parse(&value)?), + "--health-port" => health_port = Some(value.parse()?), + _ => anyhow::bail!("unknown argument: {flag}"), + } + } + + Ok(Self { + queue_url: queue_url.ok_or_else(|| anyhow::anyhow!("missing --queue-url"))?, + results_url: results_url.ok_or_else(|| anyhow::anyhow!("missing --results-url"))?, + health_port: health_port.ok_or_else(|| anyhow::anyhow!("missing --health-port"))?, + }) + } +} + +#[derive(Serialize)] +struct EmptyRequest {} + +#[derive(Deserialize)] +struct DequeueResponse { + message: Option, +} + +#[derive(Deserialize)] +struct QueueMessage { + id: u64, + payload: String, +} + +#[derive(Serialize)] +struct KvPutRequest { + value: &'static str, + expected_version: Option, +} + +#[derive(Deserialize)] +struct KvPutResponse { + applied: bool, +} diff --git a/examples/queue/queue-node/src/client.rs b/examples/queue/queue-node/src/client.rs index 1b6b4e9..b9f7704 100644 --- a/examples/queue/queue-node/src/client.rs +++ b/examples/queue/queue-node/src/client.rs @@ -16,6 +16,11 @@ impl QueueHttpClient { } } + #[must_use] + pub const fn base_url(&self) -> &Url { + &self.base_url + } + pub async fn get(&self, path: &str) -> anyhow::Result { let url = self.base_url.join(path)?; let response = self.client.get(url).send().await?.error_for_status()?; diff --git a/examples/queue/testing/integration/Cargo.toml b/examples/queue/testing/integration/Cargo.toml index c85823e..3c84c1d 100644 --- a/examples/queue/testing/integration/Cargo.toml +++ b/examples/queue/testing/integration/Cargo.toml @@ -8,6 +8,7 @@ version.workspace = true async-trait = { workspace = true } queue-node = { path = "../../queue-node" } serde = { workspace = true } +testing-framework-app = { workspace = true } testing-framework-core = { workspace = true } testing-framework-runner-compose = { workspace = true } testing-framework-runner-local = { workspace = true } diff --git a/examples/queue/testing/integration/src/app.rs b/examples/queue/testing/integration/src/app.rs index 5d42181..b324515 100644 --- a/examples/queue/testing/integration/src/app.rs +++ b/examples/queue/testing/integration/src/app.rs @@ -3,6 +3,7 @@ use std::io::Error; use async_trait::async_trait; use queue_node::QueueHttpClient; use serde::{Deserialize, Serialize}; +use testing_framework_app::{AppDeployment, AppHostEnv, DeployContext, LocalAppCluster}; use testing_framework_core::scenario::{ Application, ClusterNodeConfigApplication, ClusterNodeView, ClusterPeerView, DynError, NodeAccess, serialize_cluster_yaml_config, @@ -40,6 +41,34 @@ impl Application for QueueEnv { } } +#[derive(Clone)] +pub struct QueueLocalApp { + deployment: QueueTopology, +} + +impl QueueLocalApp { + #[must_use] + pub fn nodes(nodes: usize) -> Self { + Self { + deployment: QueueTopology::new(nodes), + } + } + + #[must_use] + pub fn deployment(&self) -> QueueTopology { + self.deployment.clone() + } +} + +#[async_trait] +impl AppDeployment for QueueLocalApp { + type Handle = LocalAppCluster; + + async fn deploy(self, ctx: &mut DeployContext) -> Result { + ctx.deploy_local_cluster::(self.deployment).await + } +} + impl ClusterNodeConfigApplication for QueueEnv { type ConfigError = Error; diff --git a/examples/queue/testing/integration/src/local_env.rs b/examples/queue/testing/integration/src/local_env.rs index 747a3a7..20018a2 100644 --- a/examples/queue/testing/integration/src/local_env.rs +++ b/examples/queue/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 QueueEnv { } fn local_process_spec() -> LocalProcessSpec { - LocalProcessSpec::new("QUEUE_NODE_BIN").with_rust_log("queue_node=info") + LocalProcessSpec::new("QUEUE_NODE_BIN") + .with_binary_provider(queue_binary_provider()) + .with_rust_log("queue_node=info") } fn render_local_config(config: &QueueNodeConfig) -> Result, DynError> { @@ -39,3 +42,31 @@ impl LocalBinaryApp for QueueEnv { config.http_port } } + +fn queue_binary_provider() -> FallbackBinaryProvider { + let workspace = workspace_root(); + let providers: [BinaryProviderRef; 2] = [ + Arc::new(EnvBinaryProvider::new("QUEUE_NODE_BIN")), + Arc::new(BuildBinaryProvider { + command: BuildCommand::new("cargo").with_args([ + "build", + "-p", + "queue-node", + "--bin", + "queue-node", + ]), + output_path: PathBuf::from(format!( + "target/debug/queue-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("../../../..") +}