From 50cc292ed2c2df53e4829ffc64582431dbb7ed7e Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Tue, 7 Jul 2026 20:57:10 -0400 Subject: [PATCH 1/4] Done implementation. --- Cargo.lock | 11 + Cargo.toml | 1 + tests/huntsman/e2e/Cargo.toml | 12 + tests/huntsman/e2e/src/lib.rs | 7 + tests/huntsman/e2e/src/test_driver.rs | 331 ++++++++++++++++++++++++++ tests/huntsman/e2e/src/types.rs | 31 +++ 6 files changed, 393 insertions(+) create mode 100644 tests/huntsman/e2e/Cargo.toml create mode 100644 tests/huntsman/e2e/src/lib.rs create mode 100644 tests/huntsman/e2e/src/test_driver.rs create mode 100644 tests/huntsman/e2e/src/types.rs diff --git a/Cargo.lock b/Cargo.lock index 326b2b189..d384e3c71 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -452,6 +452,17 @@ version = "0.15.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1aaf95b3e5c8f23aa320147307562d361db0ae0d51242340f558153b4eb2439b" +[[package]] +name = "e2e" +version = "0.1.0" +dependencies = [ + "anyhow", + "spider-client", + "spider-core", + "tokio", + "tonic", +] + [[package]] name = "either" version = "1.16.0" diff --git a/Cargo.toml b/Cargo.toml index c624e3e7e..e6ac488ba 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -14,6 +14,7 @@ members = [ "components/spider-utils", "examples/huntsman/complex/tasks", "examples/huntsman/complex/types", + "tests/huntsman/e2e", "tests/huntsman/em-runtime", "tests/huntsman/integration-test-tasks", "tests/huntsman/task-executor", diff --git a/tests/huntsman/e2e/Cargo.toml b/tests/huntsman/e2e/Cargo.toml new file mode 100644 index 000000000..28fc8f44e --- /dev/null +++ b/tests/huntsman/e2e/Cargo.toml @@ -0,0 +1,12 @@ +[package] +name = "e2e" +version = "0.1.0" +edition = "2024" +publish = false + +[dependencies] +anyhow = "1.0.98" +spider-client = { path = "../../../components/spider-client" } +spider-core = { path = "../../../components/spider-core" } +tokio = { version = "1.52.3", features = ["macros", "rt-multi-thread", "sync", "time"] } +tonic = "0.14.6" diff --git a/tests/huntsman/e2e/src/lib.rs b/tests/huntsman/e2e/src/lib.rs new file mode 100644 index 000000000..9fb2353d1 --- /dev/null +++ b/tests/huntsman/e2e/src/lib.rs @@ -0,0 +1,7 @@ +//! End-to-end integration-test harness for the huntsman suites. + +pub mod test_driver; +mod types; + +pub use test_driver::SpiderTestDriver; +pub use types::*; diff --git a/tests/huntsman/e2e/src/test_driver.rs b/tests/huntsman/e2e/src/test_driver.rs new file mode 100644 index 000000000..700cfa505 --- /dev/null +++ b/tests/huntsman/e2e/src/test_driver.rs @@ -0,0 +1,331 @@ +//! The shared end-to-end test driver that submits jobs to a live Spider deployment and drives +//! them to a terminal state. +//! +//! The driver is a lazily initialized process-wide singleton. It reads its Spider endpoint and +//! concurrency configuration from environment variables and serializes access to the underlying +//! client so that shared scenarios run concurrently up to a configured limit while exclusive +//! scenarios run in isolation. + +use std::{collections::HashMap, num::NonZeroUsize, time::Duration}; + +use anyhow::Context; +use spider_client::SpiderClient; +use spider_core::{ + job::JobState, + types::id::{JobId, ResourceGroupId}, +}; +use tokio::sync::{Mutex, OnceCell, RwLock, Semaphore}; +use tonic::transport::Endpoint; + +use crate::types::{JobSubmission, TerminationResult}; + +/// A process-wide harness for running end-to-end Spider job scenarios. +pub struct SpiderTestDriver { + client: RwLock, + concurrency_limiter: Semaphore, + resource_groups: Mutex>, +} + +impl SpiderTestDriver { + /// Runs a job scenario concurrently with other shared scenarios, up to the configured + /// concurrency limit. + /// + /// # Type Parameters + /// + /// * `OutcomeAssertionType` - The callback for asserting the terminal outcome of the job. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * [`anyhow::Error`] if the concurrency limiter has been closed. + /// * Forwards [`Self::instance`]'s return values on failure. + /// * Forwards [`Self::resolve_resource_group`]'s return values on failure. + /// * Forwards [`run_scenario`]'s return values on failure. + pub async fn run( + job_submission: JobSubmission, + timeout: Duration, + outcome_assertion: OutcomeAssertionType, + ) -> anyhow::Result<()> + where + OutcomeAssertionType: AsyncFnOnce(JobId, TerminationResult) -> anyhow::Result<()>, { + let driver = Self::instance().await?; + let client_guard = driver.client.read().await; + let _permit = driver + .concurrency_limiter + .acquire() + .await + .context("concurrency limiter closed")?; + let client = client_guard.clone(); + let resource_group_id = driver + .resolve_resource_group(&client, &job_submission.resource_group_id) + .await?; + let result = run_scenario( + client, + resource_group_id, + job_submission, + timeout, + async |_job_id: JobId| -> anyhow::Result<()> { Ok(()) }, + outcome_assertion, + ) + .await; + drop(client_guard); + result + } + + /// Runs a job scenario in exclusive isolation. + /// + /// # Type Parameters + /// + /// * `FailureInjectionType` - A background task for injecting a failure into the running job. + /// * `OutcomeAssertionType` - The callback for asserting the terminal outcome of the job. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * Forwards [`Self::instance`]'s return values on failure. + /// * Forwards [`Self::resolve_resource_group`]'s return values on failure. + /// * Forwards [`run_scenario`]'s return values on failure. + pub async fn run_exclusive( + job_submission: JobSubmission, + timeout: Duration, + failure_injection: FailureInjectionType, + outcome_assertion: OutcomeAssertionType, + ) -> anyhow::Result<()> + where + FailureInjectionType: AsyncFnOnce(JobId) -> anyhow::Result<()>, + OutcomeAssertionType: AsyncFnOnce(JobId, TerminationResult) -> anyhow::Result<()>, { + let driver = Self::instance().await?; + let client_guard = driver.client.write().await; + let client = client_guard.clone(); + let resource_group_id = driver + .resolve_resource_group(&client, &job_submission.resource_group_id) + .await?; + let result = run_scenario( + client, + resource_group_id, + job_submission, + timeout, + failure_injection, + outcome_assertion, + ) + .await; + drop(client_guard); + result + } + + /// Returns the process-wide driver instance, initializing it on first access. + /// + /// # Returns + /// + /// A reference to the process-wide driver instance on success. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * Forwards [`Self::init`]'s return values on failure. + async fn instance() -> anyhow::Result<&'static Self> { + static INSTANCE: OnceCell = OnceCell::const_new(); + INSTANCE.get_or_try_init(Self::init).await + } + + /// Initializes the driver from the environment, connecting to the configured Spider endpoint. + /// + /// # Returns + /// + /// A newly initialized driver on success. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * [`anyhow::Error`] if: + /// * The endpoint environment variable is unset. + /// * The endpoint value is not a valid Spider endpoint. + /// * Forwards [`read_concurrency`]'s return values on failure. + /// * Forwards [`SpiderClient::connect`]'s return values on failure. + async fn init() -> anyhow::Result { + const ENDPOINT_ENV_VAR: &str = "SPIDER_ENDPOINT"; + let endpoint_string = std::env::var(ENDPOINT_ENV_VAR) + .with_context(|| format!("{ENDPOINT_ENV_VAR} is not set"))?; + let endpoint = Endpoint::from_shared(endpoint_string).context("invalid spider endpoint")?; + let concurrency = read_concurrency()?; + let client = SpiderClient::connect(endpoint, concurrency).await?; + Ok(Self { + client: RwLock::new(client), + concurrency_limiter: Semaphore::new(concurrency.get()), + resource_groups: Mutex::new(HashMap::new()), + }) + } + + /// Resolves an external resource-group id to a Spider-assigned id, registering the resource + /// group on first use and caching the result. + /// + /// # Returns + /// + /// The Spider-assigned resource-group id on success. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * Forwards [`SpiderClient::add_resource_group`]'s return values on failure. + async fn resolve_resource_group( + &self, + client: &SpiderClient, + external_resource_group_id: &str, + ) -> anyhow::Result { + let mut resource_groups = self.resource_groups.lock().await; + if let Some(resource_group_id) = resource_groups.get(external_resource_group_id) { + return Ok(*resource_group_id); + } + let resource_group_id = client + .add_resource_group(external_resource_group_id.to_owned(), Vec::new()) + .await?; + resource_groups.insert(external_resource_group_id.to_owned(), resource_group_id); + drop(resource_groups); + Ok(resource_group_id) + } +} + +/// Runs a single job scenario: submits and starts the job, drives it to a terminal state while +/// concurrently running the failure injection, and forwards the outcome to the assertion. +/// +/// # Type Parameters +/// +/// * `FailureInjectionType` - A background task for injecting a failure into the running job. +/// * `OutcomeAssertionType` - The callback for asserting the terminal outcome of the job. +/// +/// # Errors +/// +/// Returns an error if: +/// +/// * [`anyhow::Error`] if the job does not reach a terminal state within `timeout`. +/// * Forwards [`submit_and_start_job`]'s return values on failure. +/// * Forwards [`poll_until_terminal`]'s return values on failure. +/// * Forwards `failure_injection`'s return values on failure. +/// * Forwards `outcome_assertion`'s return values on failure. +async fn run_scenario( + client: SpiderClient, + resource_group_id: ResourceGroupId, + job_submission: JobSubmission, + timeout: Duration, + failure_injection: FailureInjectionType, + outcome_assertion: OutcomeAssertionType, +) -> anyhow::Result<()> +where + FailureInjectionType: AsyncFnOnce(JobId) -> anyhow::Result<()>, + OutcomeAssertionType: AsyncFnOnce(JobId, TerminationResult) -> anyhow::Result<()>, { + let job_id = submit_and_start_job(&client, resource_group_id, job_submission).await?; + let result = match tokio::time::timeout(timeout, async { + let ((), termination) = tokio::try_join!( + failure_injection(job_id), + poll_until_terminal(&client, job_id), + )?; + anyhow::Result::::Ok(termination) + }) + .await + { + Ok(termination) => termination?, + Err(_elapsed) => anyhow::bail!("job did not reach a terminal state within {timeout:?}"), + }; + outcome_assertion(job_id, result).await +} + +/// Submits and starts the described job. +/// +/// # Returns +/// +/// The id of the submitted job on success. +/// +/// # Errors +/// +/// Returns an error if: +/// +/// * Forwards [`SpiderClient::submit_job`]'s return values on failure. +/// * Forwards [`SpiderClient::start_job`]'s return values on failure. +async fn submit_and_start_job( + client: &SpiderClient, + resource_group_id: ResourceGroupId, + job_submission: JobSubmission, +) -> anyhow::Result { + let job_id = client + .submit_job( + resource_group_id, + &job_submission.task_graph, + job_submission.inputs, + ) + .await?; + client.start_job(job_id).await?; + Ok(job_id) +} + +/// Polls the job's state until it reaches a terminal state. +/// +/// # Returns +/// +/// The terminal outcome of the job on success. +/// +/// # Errors +/// +/// Returns an error if: +/// +/// * Forwards [`SpiderClient::get_job_state`]'s return values on failure. +/// * Forwards [`SpiderClient::get_job_outputs`]'s return values on failure. +/// * Forwards [`SpiderClient::get_job_error`]'s return values on failure. +async fn poll_until_terminal( + client: &SpiderClient, + job_id: JobId, +) -> anyhow::Result { + const POLL_INTERVAL: Duration = Duration::from_millis(10); + + loop { + match client.get_job_state(job_id).await? { + JobState::Succeeded => { + let outputs = client.get_job_outputs(job_id).await?; + return Ok(TerminationResult::Success(outputs)); + } + JobState::Failed => { + let error_message = client.get_job_error(job_id).await?; + return Ok(TerminationResult::Failure(error_message)); + } + JobState::Cancelled => return Ok(TerminationResult::Cancelled), + _ => tokio::time::sleep(POLL_INTERVAL).await, + } + } +} + +/// Reads the configured shared-scenario concurrency from the environment, falling back to a default +/// value when unset. +/// +/// # Returns +/// +/// The configured concurrency on success. +/// +/// # Errors +/// +/// Returns an error if: +/// +/// * [`anyhow::Error`] if the concurrency value is invalid. +/// +/// # Panics +/// +/// Panics if the default value is zero. +fn read_concurrency() -> anyhow::Result { + const CONCURRENCY_ENV_VAR: &str = "SPIDER_CONCURRENCY"; + const DEFAULT_CONCURRENCY: usize = 8; + + match std::env::var(CONCURRENCY_ENV_VAR) { + Ok(value) => value.parse::().with_context(|| { + format!("{CONCURRENCY_ENV_VAR} must be a positive integer, got {value:?}") + }), + Err(std::env::VarError::NotPresent) => { + Ok(NonZeroUsize::new(DEFAULT_CONCURRENCY).expect("default concurrency is non-zero")) + } + Err(error) => { + Err(error).with_context(|| format!("{CONCURRENCY_ENV_VAR} is not valid unicode")) + } + } +} diff --git a/tests/huntsman/e2e/src/types.rs b/tests/huntsman/e2e/src/types.rs new file mode 100644 index 000000000..3cd2fa933 --- /dev/null +++ b/tests/huntsman/e2e/src/types.rs @@ -0,0 +1,31 @@ +//! Public types shared across the end-to-end test driver. + +use spider_core::{ + task::TaskGraph, + types::io::{TaskInput, TaskOutput}, +}; + +/// The terminal outcome of a job returned from the test driver. +pub enum TerminationResult { + /// The job succeeded, carrying the collected task outputs. + Success(Vec), + + /// The job failed, carrying the reported error message. + Failure(String), + + /// The job was cancelled before completion. + Cancelled, +} + +/// A description of a single job to submit through the test driver. +pub struct JobSubmission { + /// The external resource-group id the job is submitted under. The driver resolves it to a + /// Spider-assigned ID, registering it on first use. + pub resource_group_id: String, + + /// The task graph describing the job's computation. + pub task_graph: TaskGraph, + + /// The inputs supplied to the job's entry tasks. + pub inputs: Vec, +} From 26d645610c9f736af5ca85fb0a6d6a015f60100d Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Tue, 7 Jul 2026 21:18:49 -0400 Subject: [PATCH 2/4] toml lint. --- tests/huntsman/e2e/Cargo.toml | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/tests/huntsman/e2e/Cargo.toml b/tests/huntsman/e2e/Cargo.toml index 28fc8f44e..9551154ba 100644 --- a/tests/huntsman/e2e/Cargo.toml +++ b/tests/huntsman/e2e/Cargo.toml @@ -8,5 +8,8 @@ publish = false anyhow = "1.0.98" spider-client = { path = "../../../components/spider-client" } spider-core = { path = "../../../components/spider-core" } -tokio = { version = "1.52.3", features = ["macros", "rt-multi-thread", "sync", "time"] } +tokio = { + version = "1.52.3", + features = ["macros", "rt-multi-thread", "sync", "time"] +} tonic = "0.14.6" From e1aad84854340c99bfc25c6b50c644a8bd641413 Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Wed, 8 Jul 2026 17:41:16 -0400 Subject: [PATCH 3/4] Linter fix. --- tests/huntsman/e2e/src/test_driver.rs | 19 ++++++++++++------- tests/huntsman/e2e/src/types.rs | 7 +++---- 2 files changed, 15 insertions(+), 11 deletions(-) diff --git a/tests/huntsman/e2e/src/test_driver.rs b/tests/huntsman/e2e/src/test_driver.rs index 700cfa505..eb74e86da 100644 --- a/tests/huntsman/e2e/src/test_driver.rs +++ b/tests/huntsman/e2e/src/test_driver.rs @@ -6,18 +6,23 @@ //! client so that shared scenarios run concurrently up to a configured limit while exclusive //! scenarios run in isolation. -use std::{collections::HashMap, num::NonZeroUsize, time::Duration}; +use std::collections::HashMap; +use std::num::NonZeroUsize; +use std::time::Duration; use anyhow::Context; use spider_client::SpiderClient; -use spider_core::{ - job::JobState, - types::id::{JobId, ResourceGroupId}, -}; -use tokio::sync::{Mutex, OnceCell, RwLock, Semaphore}; +use spider_core::job::JobState; +use spider_core::types::id::JobId; +use spider_core::types::id::ResourceGroupId; +use tokio::sync::Mutex; +use tokio::sync::OnceCell; +use tokio::sync::RwLock; +use tokio::sync::Semaphore; use tonic::transport::Endpoint; -use crate::types::{JobSubmission, TerminationResult}; +use crate::types::JobSubmission; +use crate::types::TerminationResult; /// A process-wide harness for running end-to-end Spider job scenarios. pub struct SpiderTestDriver { diff --git a/tests/huntsman/e2e/src/types.rs b/tests/huntsman/e2e/src/types.rs index 3cd2fa933..d687c9a2b 100644 --- a/tests/huntsman/e2e/src/types.rs +++ b/tests/huntsman/e2e/src/types.rs @@ -1,9 +1,8 @@ //! Public types shared across the end-to-end test driver. -use spider_core::{ - task::TaskGraph, - types::io::{TaskInput, TaskOutput}, -}; +use spider_core::task::TaskGraph; +use spider_core::types::io::TaskInput; +use spider_core::types::io::TaskOutput; /// The terminal outcome of a job returned from the test driver. pub enum TerminationResult { From 35670139849c03607b081c0b63b1606b250d2d4b Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Thu, 9 Jul 2026 21:56:54 -0400 Subject: [PATCH 4/4] Update comment. --- tests/huntsman/e2e/src/test_driver.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/huntsman/e2e/src/test_driver.rs b/tests/huntsman/e2e/src/test_driver.rs index eb74e86da..ce8e87d8d 100644 --- a/tests/huntsman/e2e/src/test_driver.rs +++ b/tests/huntsman/e2e/src/test_driver.rs @@ -317,7 +317,8 @@ async fn poll_until_terminal( /// /// # Panics /// -/// Panics if the default value is zero. +/// Panics if the default value is zero. This shouldn't be reachable at runtime: the default number +/// is a non-zero compile-time constant. fn read_concurrency() -> anyhow::Result { const CONCURRENCY_ENV_VAR: &str = "SPIDER_CONCURRENCY"; const DEFAULT_CONCURRENCY: usize = 8;