logos-blockchain/logos-blockchain-testing
1
0
Fork 0
mirror of https://github.com/logos-blockchain/logos-blockchain-testing.git synced 2026-08-06 14:53:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(examples): add multi-app test stack

Browse Source
This commit is contained in:
andrussal 2026-07-18 12:04:25 +02:00
parent 2cba6c64c6
commit e5dc6a8d46
18 changed files with 685 additions and 319 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
Cargo.lock generated
View File

@ -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",

4
Cargo.toml
View File

@ -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",

4
examples/README.md
View File

@ -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<AppEnv>` examples are still useful for
backend-specific coverage:

5
examples/kvstore/kvstore-node/src/client.rs
View File

@ -16,6 +16,11 @@ impl KvHttpClient {
}
}
#[must_use]
pub const fn base_url(&self) -> &Url {
&self.base_url
}
pub async fn get<T: serde::de::DeserializeOwned>(&self, path: &str) -> anyhow::Result<T> {
let url = self.base_url.join(path)?;
let response = self.client.get(url).send().await?.error_for_status()?;

52
examples/multi_app/README.md
View File

@ -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::<ExampleStackHandle>()?;
let wallet = ctx.require_app::<WalletHandle>()?;
let stack = ctx.require_app::<JobStackHandle>()?;
```
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<AppEnv>` 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
```

11
examples/multi_app/e2e/Cargo.toml Normal file
View File

@ -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"] }

1
examples/multi_app/e2e/src/lib.rs Normal file
View File

@ -0,0 +1 @@
//! End-to-end coverage for composed testing-framework applications.

23
examples/multi_app/e2e/tests/local_happy_path.rs Normal file
View File

@ -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(())
}

22
examples/multi_app/examples/Cargo.toml
View File

@ -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"] }

269
examples/multi_app/examples/src/bin/typed_stack_smoke.rs
View File

@ -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<AppHostEnv> for ExampleStackApp {
type Handle = ExampleStackHandle;
async fn deploy(self, ctx: &mut DeployContext<AppHostEnv>) -> Result<Self::Handle, DynError> {
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<KvEnv>,
}
impl StoreHandle {
const fn new(cluster: LocalAppCluster<KvEnv>) -> 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<OpenRaftKvEnv>,
}
impl ConsensusHandle {
const fn new(cluster: LocalAppCluster<OpenRaftKvEnv>) -> 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<AppHostEnv> for ExampleStackWorkload {
fn name(&self) -> &str {
"multi_app_typed_stack_smoke"
}
async fn start(&self, ctx: &RunContext<AppHostEnv>) -> Result<(), DynError> {
let stack = ctx.require_app::<ExampleStackHandle>()?;
let store = ctx.require_app::<StoreHandle>()?;
let consensus = ctx.require_app::<ConsensusHandle>()?;
let wallet = ctx.require_app::<WalletHandle>()?;
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<u64>,
}
#[derive(Deserialize)]
struct KvPutResponse {
applied: bool,
}

18
examples/multi_app/fixture/Cargo.toml Normal file
View File

@ -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 }

348
examples/multi_app/fixture/src/lib.rs Normal file
View File

@ -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<AppHostEnv> for JobStackApp {
type Handle = JobStackHandle;
async fn deploy(self, ctx: &mut DeployContext<AppHostEnv>) -> Result<Self::Handle, DynError> {
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<QueueEnv>,
results: LocalAppCluster<KvEnv>,
worker: LocalProcessHandle<WorkerClient>,
}
impl JobStackHandle {
#[must_use]
pub const fn queue(&self) -> &LocalAppCluster<QueueEnv> {
&self.queue
}
#[must_use]
pub const fn results(&self) -> &LocalAppCluster<KvEnv> {
&self.results
}
#[must_use]
pub const fn worker(&self) -> &LocalProcessHandle<WorkerClient> {
&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<AppHostEnv> for JobWorkerApp {
type Handle = LocalProcessHandle<WorkerClient>;
async fn deploy(self, ctx: &mut DeployContext<AppHostEnv>) -> Result<Self::Handle, DynError> {
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<Self, DynError> {
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<AppHostEnv> for EnqueueJobs {
fn name(&self) -> &str {
"enqueue_jobs"
}
async fn start(&self, ctx: &RunContext<AppHostEnv>) -> Result<(), DynError> {
let stack = ctx.require_app::<JobStackHandle>()?;
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<AppHostEnv> for AllJobsCompleted {
fn name(&self) -> &str {
"all_jobs_completed"
}
async fn evaluate(&mut self, ctx: &RunContext<AppHostEnv>) -> Result<(), DynError> {
let stack = ctx.require_app::<JobStackHandle>()?;
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<bool, DynError> {
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<ValueRecord>,
}
#[derive(Deserialize)]
struct ValueRecord {
value: String,
}

16
examples/multi_app/job-worker/Cargo.toml Normal file
View File

@ -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"] }

125
examples/multi_app/job-worker/src/main.rs Normal file
View File

@ -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<Option<QueueMessage>> {
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<Self> {
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<QueueMessage>,
}
#[derive(Deserialize)]
struct QueueMessage {
id: u64,
payload: String,
}
#[derive(Serialize)]
struct KvPutRequest {
value: &'static str,
expected_version: Option<u64>,
}
#[derive(Deserialize)]
struct KvPutResponse {
applied: bool,
}

5
examples/queue/queue-node/src/client.rs
View File

@ -16,6 +16,11 @@ impl QueueHttpClient {
}
}
#[must_use]
pub const fn base_url(&self) -> &Url {
&self.base_url
}
pub async fn get<T: serde::de::DeserializeOwned>(&self, path: &str) -> anyhow::Result<T> {
let url = self.base_url.join(path)?;
let response = self.client.get(url).send().await?.error_for_status()?;

1
examples/queue/testing/integration/Cargo.toml
View File

@ -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 }

29
examples/queue/testing/integration/src/app.rs
View File

@ -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<AppHostEnv> for QueueLocalApp {
type Handle = LocalAppCluster<QueueEnv>;
async fn deploy(self, ctx: &mut DeployContext<AppHostEnv>) -> Result<Self::Handle, DynError> {
ctx.deploy_local_cluster::<QueueEnv>(self.deployment).await
}
}
impl ClusterNodeConfigApplication for QueueEnv {
type ConfigError = Error;

37
examples/queue/testing/integration/src/local_env.rs
View File

@ -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<Vec<u8>, 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("../../../..")
}