diff --git a/Cargo.lock b/Cargo.lock index 2aab179fa7c..293289fa483 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5361,6 +5361,7 @@ dependencies = [ "libc", "libsql", "reqwest 0.12.28", + "rig-core", "rust_decimal", "rust_decimal_macros", "serde", diff --git a/tools/ironclaw_stress/Cargo.toml b/tools/ironclaw_stress/Cargo.toml index e5d5988d62b..401d803ee05 100644 --- a/tools/ironclaw_stress/Cargo.toml +++ b/tools/ironclaw_stress/Cargo.toml @@ -41,3 +41,6 @@ reqwest = { version = "0.12", default-features = false, features = ["json", "rus tokio = { version = "1", features = ["fs", "io-util", "macros", "net", "rt-multi-thread", "sync", "time"] } tokio-postgres = { version = "0.7", optional = true, features = ["with-serde_json-1"] } uuid = { version = "1", features = ["v4", "serde"] } + +[dev-dependencies] +rig = { package = "rig-core", version = "0.33" } diff --git a/tools/ironclaw_stress/README.md b/tools/ironclaw_stress/README.md index 50d975f032e..e4294930c51 100644 --- a/tools/ironclaw_stress/README.md +++ b/tools/ironclaw_stress/README.md @@ -133,9 +133,28 @@ Use `--scenario` for a single workload. | `mixed-user-session` | Realistic user turn with configurable synthetic or provider-backed model latency. | | `context-growth` | Sequentially grows history, then loads context to expose context read amplification. | | `tool-session` | Realistic turn with synthetic tool calls, tool previews, tool results, and optional tool wait/failure paths. | +| `api-user-capacity` | End-to-end WebUI API send/read pressure against a running Reborn server. | | `cpu-burn` | Process-local CPU pressure control. | | `memory-churn` | Process-local allocation/RSS pressure control. | +`api-user-capacity` can use either pre-minted users from `--api-users-jsonl`, +a shared bearer via `--api-bearer-token`, or real per-user provisioning through +the WebUI admin surface: + +```bash +cargo run -p ironclaw_stress -- \ + --backend postgres \ + --scenario api-user-capacity \ + --api-base-url http://127.0.0.1:3900 \ + --api-admin-bearer-token "$IRONCLAW_REBORN_WEBUI_BEARER_TOKEN" \ + --api-admin-provisioners 4 \ + --users 100 \ + --concurrency 100 \ + --operations 1 \ + --api-read-qps-per-user 2 \ + --mock-llm-bind 127.0.0.1:3911 +``` + ## Presets Use `--preset` for a named single workload. Explicit CLI flags override preset diff --git a/tools/ironclaw_stress/src/api_capacity.rs b/tools/ironclaw_stress/src/api_capacity.rs index 844724cfddc..ed20514843a 100644 --- a/tools/ironclaw_stress/src/api_capacity.rs +++ b/tools/ironclaw_stress/src/api_capacity.rs @@ -3,7 +3,7 @@ use std::{ net::SocketAddr, path::Path, sync::{ - Arc, + Arc, Mutex, atomic::{AtomicBool, AtomicU64, Ordering}, }, time::{Duration, Instant}, @@ -26,9 +26,11 @@ use crate::{ }; const DEFAULT_READ_MIX: &str = "list_threads=50,timeline=45,session=5"; +const BACKGROUND_FLOW_MARKER: &str = "ironclaw-stress-background"; const MAX_ERROR_BODY_CHARS: usize = 512; const MAX_MOCK_LLM_REQUEST_BODY_BYTES: usize = 1024 * 1024; const MAX_READ_WORKER_SLEEP: Duration = Duration::from_secs(3600); +const MOCK_COMPLETION_CREATED_AT: u64 = 1_700_000_000; #[derive(Debug, Serialize, Deserialize)] pub(crate) struct ApiCapacitySummary { @@ -39,6 +41,10 @@ pub(crate) struct ApiCapacitySummary { pub(crate) read_workers: usize, pub(crate) read_mix: String, pub(crate) page_size: u32, + pub(crate) background_users: usize, + pub(crate) background_concurrency: usize, + pub(crate) background_operations_per_user: usize, + pub(crate) background_start_delay_ms: u64, pub(crate) wait_for_assistant: bool, pub(crate) terminal_timeout_ms: u64, pub(crate) poll_interval_ms: u64, @@ -47,14 +53,31 @@ pub(crate) struct ApiCapacitySummary { pub(crate) endpoints: BTreeMap, } -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Serialize, Deserialize)] pub(crate) struct MockLlmSummary { pub(crate) base_url: String, pub(crate) model: String, pub(crate) latency_ms: u64, + pub(crate) background_latency_ms: u64, pub(crate) jitter_ms: u64, pub(crate) output_bytes: usize, pub(crate) failure_rate: f64, + pub(crate) requests: u64, + pub(crate) foreground_requests: u64, + pub(crate) background_requests: u64, + pub(crate) max_in_flight: u64, + #[serde(skip_serializing_if = "Option::is_none")] + pub(crate) request_latency: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub(crate) foreground_request_latency: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub(crate) background_request_latency: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub(crate) first_request_offset_ms: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub(crate) last_request_offset_ms: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub(crate) request_start_spread_ms: Option, } #[derive(Debug, Serialize, Deserialize)] @@ -101,6 +124,24 @@ struct ApiIdentity { bearer_token: Option, } +#[derive(Debug, Serialize)] +struct ApiAdminCreateUserRequest { + email: String, + display_name: String, + role: &'static str, +} + +#[derive(Debug, Deserialize)] +struct ApiAdminCreateUserResponse { + user: ApiAdminCreatedUserRecord, + api_token: String, +} + +#[derive(Debug, Deserialize)] +struct ApiAdminCreatedUserRecord { + user_id: String, +} + #[derive(Clone)] struct ApiHarness { client: Client, @@ -144,10 +185,17 @@ struct ReadMix { #[derive(Debug)] struct MockLlmHandle { - summary: MockLlmSummary, + base_url: String, + state: Arc, stop_sender: Option>, } +impl MockLlmHandle { + fn summary(&self) -> MockLlmSummary { + self.state.summary(self.base_url.clone()) + } +} + impl Drop for MockLlmHandle { fn drop(&mut self) { if let Some(sender) = self.stop_sender.take() { @@ -161,18 +209,173 @@ struct MockLlmConfig { bind: SocketAddr, model: String, latency_ms: u64, + background_latency_ms: u64, jitter_ms: u64, output_bytes: usize, failure_rate: f64, } +#[derive(Debug)] struct MockLlmState { model: String, latency_ms: u64, + background_latency_ms: u64, jitter_ms: u64, output_bytes: usize, failure_rate: f64, counter: AtomicU64, + foreground_counter: AtomicU64, + background_counter: AtomicU64, + in_flight: AtomicU64, + max_in_flight: AtomicU64, + started_at: Instant, + request_latencies: Mutex>, + foreground_request_latencies: Mutex>, + background_request_latencies: Mutex>, + request_start_offsets: Mutex>, +} + +impl MockLlmState { + fn summary(&self, base_url: String) -> MockLlmSummary { + let request_latencies = self + .request_latencies + .lock() + .map(|latencies| latencies.clone()) + .unwrap_or_default(); + let foreground_request_latencies = self + .foreground_request_latencies + .lock() + .map(|latencies| latencies.clone()) + .unwrap_or_default(); + let background_request_latencies = self + .background_request_latencies + .lock() + .map(|latencies| latencies.clone()) + .unwrap_or_default(); + let request_start_offsets = self + .request_start_offsets + .lock() + .map(|offsets| offsets.clone()) + .unwrap_or_default(); + let first_request_offset = request_start_offsets.iter().min().copied(); + let last_request_offset = request_start_offsets.iter().max().copied(); + MockLlmSummary { + base_url, + model: self.model.clone(), + latency_ms: self.latency_ms, + background_latency_ms: self.resolved_background_latency_ms(), + jitter_ms: self.jitter_ms, + output_bytes: self.output_bytes, + failure_rate: self.failure_rate, + requests: self.counter.load(Ordering::Relaxed), + foreground_requests: self.foreground_counter.load(Ordering::Relaxed), + background_requests: self.background_counter.load(Ordering::Relaxed), + max_in_flight: self.max_in_flight.load(Ordering::Relaxed), + request_latency: duration_latency_summary(&request_latencies), + foreground_request_latency: duration_latency_summary(&foreground_request_latencies), + background_request_latency: duration_latency_summary(&background_request_latencies), + first_request_offset_ms: first_request_offset.map(|duration| duration.as_millis()), + last_request_offset_ms: last_request_offset.map(|duration| duration.as_millis()), + request_start_spread_ms: first_request_offset + .zip(last_request_offset) + .map(|(first, last)| last.saturating_sub(first).as_millis()), + } + } + + fn resolved_background_latency_ms(&self) -> u64 { + if self.background_latency_ms == 0 { + self.latency_ms + } else { + self.background_latency_ms + } + } + + fn record_request_start(&self, offset: Duration) { + if let Ok(mut offsets) = self.request_start_offsets.lock() { + offsets.push(offset); + } + } + + fn record_request_latency(&self, kind: MockRequestKind, latency: Duration) { + if let Ok(mut latencies) = self.request_latencies.lock() { + latencies.push(latency); + } + match kind { + MockRequestKind::Foreground => { + if let Ok(mut latencies) = self.foreground_request_latencies.lock() { + latencies.push(latency); + } + } + MockRequestKind::Background => { + if let Ok(mut latencies) = self.background_request_latencies.lock() { + latencies.push(latency); + } + } + } + } + + fn enter_request(&self) { + let in_flight = self.in_flight.fetch_add(1, Ordering::Relaxed) + 1; + raise_atomic_max(&self.max_in_flight, in_flight); + } + + fn exit_request(&self) { + self.in_flight.fetch_sub(1, Ordering::Relaxed); + } +} + +fn duration_latency_summary(latencies: &[Duration]) -> Option { + if latencies.is_empty() { + return None; + } + let mut latency_us = latencies + .iter() + .map(Duration::as_micros) + .collect::>(); + latency_us.sort_unstable(); + Some(latency_summary(&latency_us)) +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum MockRequestKind { + Foreground, + Background, +} + +struct MockRequestGuard { + state: Arc, + kind: MockRequestKind, + started_at: Instant, +} + +impl MockRequestGuard { + fn new(state: &Arc, kind: MockRequestKind) -> Self { + state.record_request_start(state.started_at.elapsed()); + state.enter_request(); + Self { + state: Arc::clone(state), + kind, + started_at: Instant::now(), + } + } +} + +impl Drop for MockRequestGuard { + fn drop(&mut self) { + self.state + .record_request_latency(self.kind, self.started_at.elapsed()); + self.state.exit_request(); + } +} + +fn raise_atomic_max(target: &AtomicU64, value: u64) { + let mut current = target.load(Ordering::Relaxed); + while value > current { + match target.compare_exchange_weak(current, value, Ordering::Relaxed, Ordering::Relaxed) { + Ok(_) => break, + Err(next) => current = next, + } + } } pub(crate) fn default_read_mix() -> String { @@ -190,10 +393,37 @@ pub(crate) async fn run( ) -> Result { let _mock_llm = start_mock_llm(args).await?; let harness = ApiHarness::new(args)?; - let identities = load_identities(args).await?; - let users = setup_users(args, run_id, &harness, identities).await?; + let identity_count = args.users.saturating_add(args.api_background_users).max(1); + let identities = load_identities(args, identity_count).await?; + let identities = provision_missing_identities(args, run_id, &harness, identities).await?; + let foreground_identities = select_identities(&identities, 0, args.users); + let background_identities = + select_identities(&identities, args.users, args.api_background_users); + let users = setup_users( + args, + run_id, + &harness, + foreground_identities, + args.users, + "fg", + ) + .await?; + let background_users = if args.api_background_users == 0 { + Vec::new() + } else { + setup_users( + args, + run_id, + &harness, + background_identities, + args.api_background_users, + "background", + ) + .await? + }; let read_mix = ReadMix::parse(&args.api_read_mix)?; let read_workers = read_worker_count(args); + let background_concurrency = background_concurrency_limit(args, background_users.len()); if let Some(warmup_args) = args.warmup_args() { eprintln!( @@ -210,6 +440,7 @@ pub(crate) async fn run( &warmup_namespace, &harness, Arc::new(users.clone()), + Arc::new(background_users.clone()), &read_mix, warmup_progress, ) @@ -223,17 +454,24 @@ pub(crate) async fn run( &measured_namespace, &harness, Arc::new(users), + Arc::new(background_users), &read_mix, progress, ) .await?; let elapsed = started.elapsed(); - let mock_summary = _mock_llm.as_ref().map(|handle| handle.summary.clone()); + let mock_summary = _mock_llm.as_ref().map(MockLlmHandle::summary); let mut endpoints = summarize_api_samples(&window.api_samples, elapsed); endpoints.insert( "full_flow".to_string(), summarize_flow_samples(&window.flow_samples, elapsed), ); + if !window.background_flow_samples.is_empty() { + endpoints.insert( + "background_full_flow".to_string(), + summarize_flow_samples(&window.background_flow_samples, elapsed), + ); + } let summary = ApiCapacitySummary { base_url: harness.base_url.clone(), virtual_users: args.users, @@ -242,6 +480,10 @@ pub(crate) async fn run( read_workers, read_mix: read_mix.spec, page_size: args.api_page_size, + background_users: args.api_background_users, + background_concurrency, + background_operations_per_user: args.api_background_operations, + background_start_delay_ms: args.api_background_start_delay_ms, wait_for_assistant: args.api_wait_for_assistant, terminal_timeout_ms: args.api_terminal_timeout_ms, poll_interval_ms: args.api_poll_interval_ms, @@ -259,6 +501,7 @@ pub(crate) async fn run( struct WindowResult { flow_samples: Vec, + background_flow_samples: Vec, api_samples: Vec, } @@ -294,18 +537,44 @@ async fn run_window( operation_namespace: &str, harness: &ApiHarness, users: Arc>, + background_users: Arc>, read_mix: &ReadMix, progress: Arc, ) -> Result { - let mut tasks = JoinSet::new(); + let mut background_tasks = JoinSet::new(); + let mut writer_tasks = JoinSet::new(); + let mut read_tasks = JoinSet::new(); let stop_reads = Arc::new(ReadShutdown::new()); - for user in users.iter().cloned() { + let background_limit = background_concurrency_limit(args, background_users.len()); + let mut next_background_index = 0usize; + while next_background_index < background_users.len() + && background_tasks.len() < background_limit + { + let user = background_users[next_background_index].clone(); + next_background_index += 1; + let harness = harness.clone(); + let args = args.clone(); + let operation_namespace = operation_namespace.to_string(); + background_tasks.spawn(async move { + run_background_user(&args, &operation_namespace, harness, user).await + }); + } + if !background_tasks.is_empty() && args.api_background_start_delay_ms > 0 { + tokio::time::sleep(Duration::from_millis(args.api_background_start_delay_ms)).await; + } + + let writer_limit = writer_concurrency_limit(args, users.len()); + let writer_user_count = writer_user_count_for_window(args, users.len(), writer_limit); + let mut next_writer_index = 0usize; + while next_writer_index < writer_user_count && writer_tasks.len() < writer_limit { + let user = users[next_writer_index].clone(); + next_writer_index += 1; let harness = harness.clone(); let progress = Arc::clone(&progress); let args = args.clone(); let operation_namespace = operation_namespace.to_string(); - tasks.spawn(async move { + writer_tasks.spawn(async move { run_virtual_user(&args, &operation_namespace, harness, user, progress).await }); } @@ -318,39 +587,135 @@ async fn run_window( let read_mix = read_mix.clone(); let stop_reads = Arc::clone(&stop_reads); let args = args.clone(); - tasks.spawn(async move { + read_tasks.spawn(async move { run_read_worker(&args, harness, users, read_mix, worker_index, stop_reads).await }); } } - let mut writer_remaining = users.len(); let mut flow_samples = Vec::new(); + let mut background_flow_samples = Vec::new(); let mut api_samples = Vec::new(); - while let Some(joined) = tasks.join_next().await { + while let Some(joined) = writer_tasks.join_next().await { let result = joined.map_err(|error| format!("api task join failed: {error}"))??; - if result.writer_done { - writer_remaining = writer_remaining.saturating_sub(1); - if writer_remaining == 0 { - stop_reads.stop(); - } + flow_samples.extend(result.flow_samples); + background_flow_samples.extend(result.background_flow_samples); + api_samples.extend(result.api_samples); + + if next_writer_index < writer_user_count { + let user = users[next_writer_index].clone(); + next_writer_index += 1; + let harness = harness.clone(); + let progress = Arc::clone(&progress); + let args = args.clone(); + let operation_namespace = operation_namespace.to_string(); + writer_tasks.spawn(async move { + run_virtual_user(&args, &operation_namespace, harness, user, progress).await + }); } + } + + stop_reads.stop(); + while let Some(joined) = read_tasks.join_next().await { + let result = joined.map_err(|error| format!("api read task join failed: {error}"))??; flow_samples.extend(result.flow_samples); + background_flow_samples.extend(result.background_flow_samples); + api_samples.extend(result.api_samples); + } + + while let Some(joined) = background_tasks.join_next().await { + let result = + joined.map_err(|error| format!("api background task join failed: {error}"))??; + background_flow_samples.extend(result.background_flow_samples); api_samples.extend(result.api_samples); + + if next_background_index < background_users.len() { + let user = background_users[next_background_index].clone(); + next_background_index += 1; + let harness = harness.clone(); + let args = args.clone(); + let operation_namespace = operation_namespace.to_string(); + background_tasks.spawn(async move { + run_background_user(&args, &operation_namespace, harness, user).await + }); + } } Ok(WindowResult { flow_samples, + background_flow_samples, api_samples, }) } +fn writer_concurrency_limit(args: &Args, user_count: usize) -> usize { + args.concurrency.max(1).min(user_count.max(1)) +} + +fn writer_user_count_for_window(args: &Args, user_count: usize, writer_limit: usize) -> usize { + if args.uses_duration_mode() { + writer_limit + } else { + user_count + } +} + +fn background_concurrency_limit(args: &Args, background_user_count: usize) -> usize { + if background_user_count == 0 { + 0 + } else if args.api_background_concurrency == 0 { + background_user_count.min(8) + } else { + args.api_background_concurrency + .max(1) + .min(background_user_count) + } +} + struct TaskResult { - writer_done: bool, flow_samples: Vec, + background_flow_samples: Vec, api_samples: Vec, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ApiFlowKind { + Foreground, + Background, +} + +impl ApiFlowKind { + fn send_sample_name(self) -> &'static str { + match self { + Self::Foreground => "send_message", + Self::Background => "background_send_message", + } + } + + fn timeline_sample_name(self) -> &'static str { + match self { + Self::Foreground => "timeline", + Self::Background => "background_timeline", + } + } + + fn wait_sample_name(self) -> &'static str { + match self { + Self::Foreground => "wait_for_assistant", + Self::Background => "background_wait_for_assistant", + } + } + + fn payload_prefix(self, operation_ref: &str) -> String { + match self { + Self::Foreground => format!("api stress message {operation_ref}"), + Self::Background => { + format!("{BACKGROUND_FLOW_MARKER} api stress background message {operation_ref}") + } + } + } +} + async fn run_virtual_user( args: &Args, operation_namespace: &str, @@ -361,6 +726,7 @@ async fn run_virtual_user( let target = args.operation_target(); let started = Instant::now(); let mut operation_index = 0; + let mut expected_finalized_assistant_count = 0usize; let mut flow_samples = Vec::with_capacity(args.initial_worker_sample_capacity()); let mut api_samples = Vec::new(); @@ -369,8 +735,17 @@ async fn run_virtual_user( "{operation_namespace}:{}:{}:{}", user.label, user.index, operation_index ); - let (flow, mut operation_api_samples) = - run_full_flow(args, &harness, &user, operation_index, &operation_ref).await; + let (flow, mut operation_api_samples, updated_finalized_assistant_count) = run_full_flow( + args, + &harness, + &user, + operation_index, + &operation_ref, + expected_finalized_assistant_count, + ApiFlowKind::Foreground, + ) + .await; + expected_finalized_assistant_count = updated_finalized_assistant_count; progress.record(flow.error.is_some(), flow.latency); api_samples.append(&mut operation_api_samples); flow_samples.push(flow); @@ -383,8 +758,45 @@ async fn run_virtual_user( } Ok(TaskResult { - writer_done: true, flow_samples, + background_flow_samples: Vec::new(), + api_samples, + }) +} + +async fn run_background_user( + args: &Args, + operation_namespace: &str, + harness: ApiHarness, + user: ApiUser, +) -> Result { + let mut expected_finalized_assistant_count = 0usize; + let mut background_flow_samples = Vec::with_capacity(args.api_background_operations); + let mut api_samples = Vec::new(); + + for operation_index in 0..args.api_background_operations { + let operation_ref = format!( + "{operation_namespace}:background:{}:{}:{}", + user.label, user.index, operation_index + ); + let (flow, mut operation_api_samples, updated_finalized_assistant_count) = run_full_flow( + args, + &harness, + &user, + operation_index, + &operation_ref, + expected_finalized_assistant_count, + ApiFlowKind::Background, + ) + .await; + expected_finalized_assistant_count = updated_finalized_assistant_count; + api_samples.append(&mut operation_api_samples); + background_flow_samples.push(flow); + } + + Ok(TaskResult { + flow_samples: Vec::new(), + background_flow_samples, api_samples, }) } @@ -395,20 +807,22 @@ async fn run_full_flow( user: &ApiUser, operation_index: usize, operation_ref: &str, -) -> (Sample, Vec) { + expected_finalized_assistant_count: usize, + flow_kind: ApiFlowKind, +) -> (Sample, Vec, usize) { let started = Instant::now(); let mut api_samples = Vec::new(); let body = json!({ "client_action_id": format!("ironclaw-stress-api:{operation_ref}"), "content": stress_payload( - format!("api stress message {operation_ref}"), + flow_kind.payload_prefix(operation_ref), args.user_message_bytes, ), }); let send = harness .request_json( user, - "send_message", + flow_kind.send_sample_name(), Method::POST, &format!("/api/webchat/v2/threads/{}/messages", user.thread_id), Some(body), @@ -418,14 +832,69 @@ async fn run_full_flow( api_samples.push(send.sample); let failure = match send_value { Ok(value) => { + let target_finalized_count = expected_finalized_assistant_count + 1; if args.api_wait_for_assistant { - wait_for_assistant(args, harness, user, operation_index, &mut api_samples).await + let wait_started = Instant::now(); + let wait = wait_for_assistant( + args, + harness, + user, + operation_index, + target_finalized_count, + flow_kind, + &mut api_samples, + ) + .await; + api_samples.push(ApiRequestSample { + name: flow_kind.wait_sample_name(), + latency: wait_started.elapsed(), + failure: wait.as_ref().err().cloned(), + }); + match wait { + Ok(finalized_count) => { + let latency = started.elapsed(); + return ( + Sample { + latency, + error: None, + failure: None, + stages: None, + }, + api_samples, + finalized_count.max(target_finalized_count), + ); + } + Err(failure) => { + let latency = started.elapsed(); + let error = Some(failure.bucket.clone()); + return ( + Sample { + latency, + error, + failure: Some(failure), + stages: None, + }, + api_samples, + target_finalized_count, + ); + } + } } else if submitted_or_already_submitted(&value) { - None + let latency = started.elapsed(); + return ( + Sample { + latency, + error: None, + failure: None, + stages: None, + }, + api_samples, + target_finalized_count, + ); } else { Some(FailureCause::new( "api_submit_not_accepted", - "send_message", + flow_kind.send_sample_name(), compact_json(&value), )) } @@ -443,6 +912,7 @@ async fn run_full_flow( stages: None, }, api_samples, + expected_finalized_assistant_count, ) } @@ -451,23 +921,28 @@ async fn wait_for_assistant( harness: &ApiHarness, user: &ApiUser, operation_index: usize, + target_finalized_count: usize, + flow_kind: ApiFlowKind, api_samples: &mut Vec, -) -> Option { +) -> Result { let deadline = Instant::now() + Duration::from_millis(args.api_terminal_timeout_ms); loop { - let timeline = harness.timeline(user).await; + let timeline = harness + .timeline_with_name(user, flow_kind.timeline_sample_name()) + .await; let value = timeline.value.clone(); api_samples.push(timeline.sample); match value { Ok(value) => { - if timeline_has_finalized_assistant(&value) { - return None; + let finalized_count = timeline_finalized_assistant_count(&value); + if finalized_count >= target_finalized_count { + return Ok(finalized_count); } } - Err(failure) => return Some(failure), + Err(failure) => return Err(failure), } if Instant::now() >= deadline { - return Some(FailureCause::new( + return Err(FailureCause::new( "api_full_flow_timeout", "timeline", format!( @@ -520,8 +995,8 @@ async fn run_read_worker( } Ok(TaskResult { - writer_done: false, flow_samples: Vec::new(), + background_flow_samples: Vec::new(), api_samples, }) } @@ -573,6 +1048,44 @@ impl ApiHarness { .await } + async fn create_admin_user( + &self, + admin_bearer_token: &str, + email: String, + display_name: String, + role: &'static str, + ) -> ApiCallResult { + let request = ApiAdminCreateUserRequest { + email, + display_name, + role, + }; + let body = match serde_json::to_value(request) { + Ok(value) => value, + Err(error) => { + let detail = error.to_string(); + let failure = + FailureCause::new("api_json_encode", "admin_create_user", detail.clone()); + return ApiCallResult { + sample: ApiRequestSample { + name: "admin_create_user", + latency: Duration::ZERO, + failure: Some(failure.clone()), + }, + value: Err(failure), + }; + } + }; + self.request_json_with_bearer( + Some(admin_bearer_token), + "admin_create_user", + Method::POST, + "/api/webchat/v2/admin/users", + Some(body), + ) + .await + } + async fn list_threads(&self, user: &ApiUser) -> ApiCallResult { self.request_json( user, @@ -585,9 +1098,13 @@ impl ApiHarness { } async fn timeline(&self, user: &ApiUser) -> ApiCallResult { + self.timeline_with_name(user, "timeline").await + } + + async fn timeline_with_name(&self, user: &ApiUser, name: &'static str) -> ApiCallResult { self.request_json( user, - "timeline", + name, Method::GET, &format!( "/api/webchat/v2/threads/{}/timeline?limit={}", @@ -616,11 +1133,23 @@ impl ApiHarness { method: Method, path: &str, body: Option, + ) -> ApiCallResult { + self.request_json_with_bearer(user.bearer_token.as_deref(), name, method, path, body) + .await + } + + async fn request_json_with_bearer( + &self, + bearer_token: Option<&str>, + name: &'static str, + method: Method, + path: &str, + body: Option, ) -> ApiCallResult { let url = format!("{}{}", self.base_url, path); let started = Instant::now(); let mut request = self.client.request(method, &url); - if let Some(token) = &user.bearer_token { + if let Some(token) = bearer_token { request = request.bearer_auth(token); } if let Some(body) = body { @@ -681,17 +1210,19 @@ async fn setup_users( run_id: &str, harness: &ApiHarness, identities: Vec, + user_count: usize, + thread_label: &str, ) -> Result, String> { let setup_concurrency = args.api_setup_concurrency.max(1); let mut next_index = 0; let mut join_set = JoinSet::new(); - let mut users = Vec::with_capacity(args.users); + let mut users = Vec::with_capacity(user_count); - while next_index < args.users || !join_set.is_empty() { - while next_index < args.users && join_set.len() < setup_concurrency { + while next_index < user_count || !join_set.is_empty() { + while next_index < user_count && join_set.len() < setup_concurrency { let harness = harness.clone(); let identity = identities[next_index % identities.len()].clone(); - let thread_id = format!("stress-{run_id}-{next_index}"); + let thread_id = format!("stress-{run_id}-{thread_label}-{next_index}"); let user_index = next_index; join_set.spawn(async move { let create = harness.create_thread(&identity, &thread_id).await; @@ -723,18 +1254,250 @@ async fn setup_users( Ok(users) } -async fn load_identities(args: &Args) -> Result, String> { +fn select_identities( + identities: &[ApiIdentity], + start_index: usize, + count: usize, +) -> Vec { + if count == 0 { + return Vec::new(); + } + (0..count) + .map(|offset| identities[(start_index + offset) % identities.len()].clone()) + .collect() +} + +async fn provision_missing_identities( + args: &Args, + run_id: &str, + harness: &ApiHarness, + identities: Vec, +) -> Result, String> { + let Some(admin_bearer_token) = args.api_admin_bearer_token.as_deref() else { + return Ok(identities); + }; + if admin_bearer_token.trim().is_empty() { + return Err("--api-admin-bearer-token must not be empty".to_string()); + } + let missing_tokens = identities + .iter() + .filter(|identity| identity.bearer_token.is_none()) + .count(); + if missing_tokens == 0 { + return Ok(identities); + } + + eprintln!( + "{} provisioning {missing_tokens} API users through admin CRUD", + crate::log_prefix(args) + ); + let provisioner_tokens = + prepare_admin_provisioners(args, run_id, harness, admin_bearer_token).await?; + + let setup_concurrency = args.api_setup_concurrency.max(1); + let mut results = vec![None; identities.len()]; + let mut next_index = 0; + let mut join_set = JoinSet::new(); + + while next_index < identities.len() || !join_set.is_empty() { + while next_index < identities.len() && join_set.len() < setup_concurrency { + let identity = identities[next_index].clone(); + let identity_index = next_index; + next_index += 1; + + if identity.bearer_token.is_some() { + results[identity_index] = Some(identity); + continue; + } + + let harness = harness.clone(); + let provisioner_token = + provisioner_tokens[identity_index % provisioner_tokens.len()].clone(); + let email = admin_user_email(run_id, identity_index); + let display_name = format!("IronClaw Stress {}", identity.label); + join_set.spawn(async move { + let create = harness + .create_admin_user(&provisioner_token, email, display_name, "member") + .await; + match create.value { + Ok(value) => { + let created: ApiAdminCreateUserResponse = + serde_json::from_value(value).map_err(|error| { + format!("parse admin create response for api identity {identity_index}: {error}") + })?; + if created.api_token.is_empty() { + return Err(format!( + "admin create response for api identity {identity_index} returned an empty api_token" + )); + } + Ok(( + identity_index, + ApiIdentity { + label: created.user.user_id, + bearer_token: Some(created.api_token), + }, + )) + } + Err(failure) => Err(format!( + "admin create user for api identity {identity_index}: {}: {}", + failure.bucket, failure.detail + )), + } + }); + } + + if join_set.is_empty() { + continue; + } + + let (identity_index, identity) = join_set + .join_next() + .await + .ok_or_else(|| "admin provisioning join set ended unexpectedly".to_string())? + .map_err(|error| format!("admin provisioning task join failed: {error}"))??; + results[identity_index] = Some(identity); + } + + results + .into_iter() + .enumerate() + .map(|(index, identity)| { + identity.ok_or_else(|| format!("api identity {index} was not initialized")) + }) + .collect() +} + +async fn prepare_admin_provisioners( + args: &Args, + run_id: &str, + harness: &ApiHarness, + admin_bearer_token: &str, +) -> Result, String> { + if args.api_admin_provisioners <= 1 { + return Ok(vec![admin_bearer_token.to_string()]); + } + + eprintln!( + "{} creating {} admin provisioning principals", + crate::log_prefix(args), + args.api_admin_provisioners + ); + + let setup_concurrency = args.api_setup_concurrency.max(1); + let mut next_index = 0; + let mut join_set = JoinSet::new(); + let mut tokens = vec![None; args.api_admin_provisioners]; + + while next_index < args.api_admin_provisioners || !join_set.is_empty() { + while next_index < args.api_admin_provisioners && join_set.len() < setup_concurrency { + let provisioner_index = next_index; + next_index += 1; + + let harness = harness.clone(); + let admin_bearer_token = admin_bearer_token.to_string(); + let provisioner_run_id = format!("{run_id}-admin-provisioner"); + let email = admin_user_email(&provisioner_run_id, provisioner_index); + let display_name = format!("IronClaw Stress Provisioner {provisioner_index}"); + join_set.spawn(async move { + let create = harness + .create_admin_user(&admin_bearer_token, email, display_name, "admin") + .await; + match create.value { + Ok(value) => { + let created: ApiAdminCreateUserResponse = + serde_json::from_value(value).map_err(|error| { + format!( + "parse admin provisioner response {provisioner_index}: {error}" + ) + })?; + if created.api_token.is_empty() { + return Err(format!( + "admin provisioner response {provisioner_index} returned an empty api_token" + )); + } + Ok((provisioner_index, created.api_token)) + } + Err(failure) => Err(format!( + "create admin provisioner {provisioner_index}: {}: {}", + failure.bucket, failure.detail + )), + } + }); + } + + let (provisioner_index, token) = join_set + .join_next() + .await + .ok_or_else(|| "admin provisioner join set ended unexpectedly".to_string())? + .map_err(|error| format!("admin provisioner task join failed: {error}"))??; + tokens[provisioner_index] = Some(token); + } + + tokens + .into_iter() + .enumerate() + .map(|(index, token)| { + token.ok_or_else(|| format!("admin provisioner {index} was not initialized")) + }) + .collect() +} + +async fn load_identities(args: &Args, identity_count: usize) -> Result, String> { if let Some(path) = &args.api_users_jsonl { return load_identity_file(path).await; } - Ok((0..args.users) + let bearer_token = generated_identity_bearer_token(args); + Ok((0..identity_count) .map(|index| ApiIdentity { label: format!("user-{index}"), - bearer_token: args.api_bearer_token.clone(), + bearer_token: bearer_token.clone(), }) .collect()) } +fn generated_identity_bearer_token(args: &Args) -> Option { + if args.api_admin_bearer_token.is_some() { + None + } else { + args.api_bearer_token.clone() + } +} + +fn admin_user_email(run_id: &str, index: usize) -> String { + let run_slug = sanitize_email_local_part(run_id); + let mut local = if run_slug.is_empty() { + format!("stress-{index}") + } else { + format!("stress-{index}-{run_slug}") + }; + local.truncate(64); + while local.ends_with('-') { + local.pop(); + } + if local.is_empty() { + local.push_str("stress"); + } + format!("{local}@ironclaw-stress.local") +} + +fn sanitize_email_local_part(value: &str) -> String { + let mut sanitized = String::with_capacity(value.len().min(64)); + let mut previous_dash = false; + for ch in value.chars() { + let next = if ch.is_ascii_alphanumeric() { + previous_dash = false; + ch.to_ascii_lowercase() + } else if previous_dash { + continue; + } else { + previous_dash = true; + '-' + }; + sanitized.push(next); + } + sanitized.trim_matches('-').to_string() +} + async fn load_identity_file(path: &Path) -> Result, String> { let contents = tokio::fs::read_to_string(path) .await @@ -957,17 +1720,18 @@ fn submitted_or_already_submitted(value: &Value) -> bool { ) } -fn timeline_has_finalized_assistant(value: &Value) -> bool { +fn timeline_finalized_assistant_count(value: &Value) -> usize { value .get("messages") .and_then(Value::as_array) .into_iter() .flatten() - .any(|message| { + .filter(|message| { let kind = message.get("kind").and_then(Value::as_str); let status = message.get("status").and_then(Value::as_str); - kind == Some("assistant") && matches!(status, Some("finalized" | "submitted")) + kind == Some("assistant") && status == Some("finalized") }) + .count() } fn stress_payload(prefix: String, minimum_bytes: usize) -> String { @@ -1006,6 +1770,7 @@ async fn start_mock_llm(args: &Args) -> Result, String> { bind, model: args.mock_llm_model.clone(), latency_ms: args.mock_llm_latency_ms, + background_latency_ms: args.mock_llm_background_latency_ms, jitter_ms: args.mock_llm_jitter_ms, output_bytes: args.mock_llm_output_bytes, failure_rate: args.mock_llm_failure_rate, @@ -1019,12 +1784,23 @@ async fn start_mock_llm(args: &Args) -> Result, String> { let state = Arc::new(MockLlmState { model: config.model.clone(), latency_ms: config.latency_ms, + background_latency_ms: config.background_latency_ms, jitter_ms: config.jitter_ms, output_bytes: config.output_bytes, failure_rate: config.failure_rate, counter: AtomicU64::new(0), + foreground_counter: AtomicU64::new(0), + background_counter: AtomicU64::new(0), + in_flight: AtomicU64::new(0), + max_in_flight: AtomicU64::new(0), + started_at: Instant::now(), + request_latencies: Mutex::new(Vec::new()), + foreground_request_latencies: Mutex::new(Vec::new()), + background_request_latencies: Mutex::new(Vec::new()), + request_start_offsets: Mutex::new(Vec::new()), }); let (stop_sender, mut stop_receiver) = oneshot::channel(); + let accept_state = Arc::clone(&state); tokio::spawn(async move { loop { tokio::select! { @@ -1032,7 +1808,7 @@ async fn start_mock_llm(args: &Args) -> Result, String> { accepted = listener.accept() => { match accepted { Ok((stream, _)) => { - let state = Arc::clone(&state); + let state = Arc::clone(&accept_state); tokio::spawn(async move { let _ = handle_mock_llm_connection(stream, state).await; }); @@ -1053,14 +1829,8 @@ async fn start_mock_llm(args: &Args) -> Result, String> { config.model ); Ok(Some(MockLlmHandle { - summary: MockLlmSummary { - base_url, - model: config.model, - latency_ms: config.latency_ms, - jitter_ms: config.jitter_ms, - output_bytes: config.output_bytes, - failure_rate: config.failure_rate, - }, + base_url, + state, stop_sender: Some(stop_sender), })) } @@ -1137,7 +1907,17 @@ async fn handle_mock_completion( body: &[u8], state: Arc, ) -> Result<(), String> { + let request_kind = classify_mock_request(body); + let _request_guard = MockRequestGuard::new(&state, request_kind); let request_index = state.counter.fetch_add(1, Ordering::Relaxed) + 1; + match request_kind { + MockRequestKind::Foreground => { + state.foreground_counter.fetch_add(1, Ordering::Relaxed); + } + MockRequestKind::Background => { + state.background_counter.fetch_add(1, Ordering::Relaxed); + } + } if should_fail_mock_request(state.failure_rate, request_index) { write_mock_response( stream, @@ -1153,7 +1933,11 @@ async fn handle_mock_completion( } else { deterministic_jitter_ms(request_index, state.jitter_ms) }; - let wait = state.latency_ms.saturating_add(jitter); + let base_latency_ms = match request_kind { + MockRequestKind::Foreground => state.latency_ms, + MockRequestKind::Background => state.resolved_background_latency_ms(), + }; + let wait = base_latency_ms.saturating_add(jitter); if wait > 0 { tokio::time::sleep(Duration::from_millis(wait)).await; } @@ -1164,37 +1948,86 @@ async fn handle_mock_completion( .unwrap_or(false); let content = stress_payload("mock assistant response".to_string(), state.output_bytes); if stream_response { - let chunk = json!({ - "id": format!("chatcmpl-stress-{request_index}"), - "object": "chat.completion.chunk", - "choices": [{"index": 0, "delta": {"content": content}, "finish_reason": null}] - }); - let done = json!({ - "id": format!("chatcmpl-stress-{request_index}"), - "object": "chat.completion.chunk", - "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}] - }); + let (chunk, done) = mock_streaming_completion_chunks(&state.model, request_index, content); let payload = format!("data: {chunk}\n\ndata: {done}\n\ndata: [DONE]\n\n"); write_mock_response(stream, 200, "text/event-stream", payload.as_bytes()).await } else { - let response = json!({ - "id": format!("chatcmpl-stress-{request_index}"), - "object": "chat.completion", - "choices": [{ - "index": 0, - "message": {"role": "assistant", "content": content}, - "finish_reason": "stop" - }], - "usage": { - "prompt_tokens": 16, - "completion_tokens": 16, - "total_tokens": 32 - } - }); + let response = mock_completion_response(&state.model, request_index, content); write_json_response(stream, 200, &response).await } } +fn classify_mock_request(body: &[u8]) -> MockRequestKind { + if std::str::from_utf8(body).is_ok_and(|body| body.contains(BACKGROUND_FLOW_MARKER)) { + MockRequestKind::Background + } else { + MockRequestKind::Foreground + } +} + +fn mock_completion_response(model: &str, request_index: u64, content: String) -> Value { + json!({ + "id": format!("chatcmpl-stress-{request_index}"), + "object": "chat.completion", + "created": MOCK_COMPLETION_CREATED_AT, + "model": model, + "system_fingerprint": null, + "choices": [{ + "index": 0, + "message": { + "role": "assistant", + "content": content, + "tool_calls": null, + "refusal": null + }, + "logprobs": null, + "finish_reason": "stop" + }], + "usage": { + "prompt_tokens": 16, + "completion_tokens": 16, + "total_tokens": 32, + "prompt_tokens_details": null + } + }) +} + +fn mock_streaming_completion_chunks( + model: &str, + request_index: u64, + content: String, +) -> (Value, Value) { + let base = json!({ + "id": format!("chatcmpl-stress-{request_index}"), + "object": "chat.completion.chunk", + "created": MOCK_COMPLETION_CREATED_AT, + "model": model, + "system_fingerprint": null, + }); + let mut chunk = base.clone(); + chunk["choices"] = json!([{ + "index": 0, + "delta": {"content": content}, + "logprobs": null, + "finish_reason": null + }]); + + let mut done = base; + done["choices"] = json!([{ + "index": 0, + "delta": {}, + "logprobs": null, + "finish_reason": "stop" + }]); + done["usage"] = json!({ + "prompt_tokens": 16, + "completion_tokens": 16, + "total_tokens": 32, + "prompt_tokens_details": null + }); + (chunk, done) +} + fn should_fail_mock_request(failure_rate: f64, request_index: u64) -> bool { if failure_rate <= 0.0 { return false; @@ -1266,6 +2099,21 @@ async fn write_mock_response( #[cfg(test)] mod tests { use super::*; + use clap::Parser; + + fn parsed_api_args(extra: &[&str]) -> Args { + let mut args = vec![ + "ironclaw_stress", + "--backend", + "libsql", + "--scenario", + "api-user-capacity", + "--api-base-url", + "http://127.0.0.1:4216", + ]; + args.extend_from_slice(extra); + Args::parse_from(args) + } #[test] fn read_mix_parses_aliases_and_weights() { @@ -1282,6 +2130,27 @@ mod tests { assert_eq!(extract_thread_id(&value).as_deref(), Some("thread-1")); } + #[test] + fn admin_user_email_is_stable_ascii_and_bounded() { + let email = admin_user_email("Run 01 / With Symbols", 42); + assert_eq!(email, "stress-42-run-01-with-symbols@ironclaw-stress.local"); + let long = admin_user_email(&"x".repeat(200), 7); + let local = long.split_once('@').unwrap().0; + assert!(local.len() <= 64); + assert!(long.ends_with("@ironclaw-stress.local")); + } + + #[test] + fn parses_admin_create_user_token_response() { + let created: ApiAdminCreateUserResponse = serde_json::from_value(json!({ + "user": {"user_id": "stress-user"}, + "api_token": "session-token" + })) + .expect("admin create response parses"); + assert_eq!(created.user.user_id, "stress-user"); + assert_eq!(created.api_token, "session-token"); + } + #[test] fn detects_finalized_assistant_in_timeline() { let value = json!({ @@ -1290,7 +2159,30 @@ mod tests { {"kind": "assistant", "status": "finalized"} ] }); - assert!(timeline_has_finalized_assistant(&value)); + assert_eq!(timeline_finalized_assistant_count(&value), 1); + } + + #[test] + fn counts_only_finalized_assistant_messages_in_timeline() { + let value = json!({ + "messages": [ + {"kind": "assistant", "status": "submitted"}, + {"kind": "assistant", "status": "finalized"}, + {"kind": "user", "status": "finalized"}, + {"kind": "assistant", "status": "finalized"} + ] + }); + assert_eq!(timeline_finalized_assistant_count(&value), 2); + } + + #[test] + fn submitted_assistant_is_not_finalized_for_full_flow() { + let value = json!({ + "messages": [ + {"kind": "assistant", "status": "submitted"} + ] + }); + assert_eq!(timeline_finalized_assistant_count(&value), 0); } #[test] @@ -1299,6 +2191,14 @@ mod tests { assert_eq!(deterministic_jitter_ms(42, u64::MAX), expected); } + #[test] + fn mock_completion_response_matches_rig_openai_shape() { + let response = mock_completion_response("stress-mock", 7, "ok".to_string()); + + serde_json::from_value::(response) + .expect("stress mock should deserialize as rig-core OpenAI completion response"); + } + #[test] fn read_worker_interval_clamps_invalid_or_huge_values() { assert_eq!( @@ -1314,4 +2214,58 @@ mod tests { MAX_READ_WORKER_SLEEP ); } + + #[test] + fn fixed_api_window_processes_all_users_but_caps_active_writers() { + let args = parsed_api_args(&["--users", "100", "--concurrency", "1", "--operations", "1"]); + let writer_limit = writer_concurrency_limit(&args, args.users); + + assert_eq!(writer_limit, 1); + assert_eq!( + writer_user_count_for_window(&args, args.users, writer_limit), + 100 + ); + } + + #[test] + fn duration_api_window_uses_active_writer_set_only() { + let args = parsed_api_args(&[ + "--users", + "100", + "--concurrency", + "10", + "--duration-seconds", + "60", + ]); + let writer_limit = writer_concurrency_limit(&args, args.users); + + assert_eq!(writer_limit, 10); + assert_eq!( + writer_user_count_for_window(&args, args.users, writer_limit), + 10 + ); + } + + #[test] + fn background_concurrency_auto_caps_at_trigger_default_shape() { + let args = parsed_api_args(&["--api-background-users", "100"]); + + assert_eq!(background_concurrency_limit(&args, 100), 8); + assert_eq!(background_concurrency_limit(&args, 4), 4); + assert_eq!(background_concurrency_limit(&args, 0), 0); + } + + #[test] + fn mock_request_classifier_detects_background_marker() { + assert_eq!( + classify_mock_request(br#"{"messages":[{"content":"hello"}]}"#), + MockRequestKind::Foreground + ); + assert_eq!( + classify_mock_request( + br#"{"messages":[{"content":"ironclaw-stress-background long task"}]}"# + ), + MockRequestKind::Background + ); + } } diff --git a/tools/ironclaw_stress/src/main.rs b/tools/ironclaw_stress/src/main.rs index e36127b56e6..fde1261a64d 100644 --- a/tools/ironclaw_stress/src/main.rs +++ b/tools/ironclaw_stress/src/main.rs @@ -197,6 +197,14 @@ pub(crate) struct Args { #[arg(long)] pub(crate) api_bearer_token: Option, + /// Operator/admin bearer token used to provision API test users via WebUI admin CRUD. + #[arg(long)] + pub(crate) api_admin_bearer_token: Option, + + /// Admin principals used to fan out API test-user provisioning. 1 uses the supplied admin token directly. + #[arg(long, default_value_t = 1)] + pub(crate) api_admin_provisioners: usize, + /// Aggregate read pressure per virtual user for api-user-capacity. #[arg(long, default_value_t = 0.0)] pub(crate) api_read_qps_per_user: f64, @@ -237,6 +245,22 @@ pub(crate) struct Args { #[arg(long, default_value_t = 16)] pub(crate) api_setup_concurrency: usize, + /// Background long-running API users to run alongside foreground api-user-capacity sends. + #[arg(long, default_value_t = 0)] + pub(crate) api_background_users: usize, + + /// Max concurrently active background API users. 0 auto-scales from --api-background-users. + #[arg(long, default_value_t = 0)] + pub(crate) api_background_concurrency: usize, + + /// Full-flow operations per background API user. + #[arg(long, default_value_t = 1)] + pub(crate) api_background_operations: usize, + + /// Delay after launching background users before foreground sends start. + #[arg(long, default_value_t = 250)] + pub(crate) api_background_start_delay_ms: u64, + /// Optional bind address for the built-in OpenAI-compatible mock LLM sidecar. #[arg(long)] pub(crate) mock_llm_bind: Option, @@ -249,6 +273,10 @@ pub(crate) struct Args { #[arg(long, default_value_t = 0)] pub(crate) mock_llm_latency_ms: u64, + /// Latency for mock LLM requests tagged as API background load. 0 uses --mock-llm-latency-ms. + #[arg(long, default_value_t = 0)] + pub(crate) mock_llm_background_latency_ms: u64, + /// Deterministic jitter ceiling added by the built-in mock LLM. #[arg(long, default_value_t = 0)] pub(crate) mock_llm_jitter_ms: u64, @@ -1087,6 +1115,10 @@ fn validate_args(args: &Args) -> Result<(), String> { if args.processes > 1 { return Err("--scenario api-user-capacity requires --processes 1".to_string()); } + } else if args.api_admin_bearer_token.is_some() { + return Err( + "--api-admin-bearer-token is only valid for --scenario api-user-capacity".to_string(), + ); } if args.api_read_qps_per_user < 0.0 { return Err("--api-read-qps-per-user must be greater than or equal to 0".to_string()); @@ -1106,6 +1138,15 @@ fn validate_args(args: &Args) -> Result<(), String> { if args.api_setup_concurrency == 0 { return Err("--api-setup-concurrency must be greater than 0".to_string()); } + if args.api_admin_provisioners == 0 { + return Err("--api-admin-provisioners must be greater than 0".to_string()); + } + if args.api_background_users > 0 && args.api_background_operations == 0 { + return Err( + "--api-background-operations must be greater than 0 when background users are enabled" + .to_string(), + ); + } api_capacity::validate_read_mix(&args.api_read_mix)?; if !(0.0..=1.0).contains(&args.mock_llm_failure_rate) { return Err("--mock-llm-failure-rate must be between 0.0 and 1.0".to_string()); @@ -1712,13 +1753,22 @@ async fn run_api_capacity_in_process( OperationTarget::Fixed { .. } => Some(args.users.saturating_mul(args.operations)), OperationTarget::Duration { .. } => None, }; + let workload_label = match operation_target { + OperationTarget::Fixed { .. } => { + format!( + "total_operations={}", + args.users.saturating_mul(args.operations) + ) + } + OperationTarget::Duration { .. } => operation_target.label(), + }; eprintln!( "{} running target={} virtual_users={} operations_per_user={} {} warmup_seconds={} read_qps_per_user={:.2} progress_interval_seconds={}", log_prefix(args), target, args.users, args.operations, - operation_target.label(), + workload_label, args.warmup_seconds, args.api_read_qps_per_user, args.progress_interval_seconds diff --git a/tools/ironclaw_stress/src/tests.rs b/tools/ironclaw_stress/src/tests.rs index e72dabc4c36..bef8c37e055 100644 --- a/tools/ironclaw_stress/src/tests.rs +++ b/tools/ironclaw_stress/src/tests.rs @@ -1216,6 +1216,8 @@ fn test_args() -> Args { api_base_url: None, api_users_jsonl: None, api_bearer_token: None, + api_admin_bearer_token: None, + api_admin_provisioners: 1, api_read_qps_per_user: 0.0, api_read_workers: 0, api_read_mix: crate::api_capacity::default_read_mix(), @@ -1226,9 +1228,14 @@ fn test_args() -> Args { api_poll_interval_ms: 250, api_request_timeout_ms: 10_000, api_setup_concurrency: 16, + api_background_users: 0, + api_background_concurrency: 0, + api_background_operations: 1, + api_background_start_delay_ms: 250, mock_llm_bind: None, mock_llm_model: "stress-mock".to_string(), mock_llm_latency_ms: 0, + mock_llm_background_latency_ms: 0, mock_llm_jitter_ms: 0, mock_llm_output_bytes: 128, mock_llm_failure_rate: 0.0,