11b2215664
**Most Efficient Warning Cleanup** (5 agents, sequential phases, 2-3 hours) ## Summary Eliminated 2421 of 2484 compilation warnings (97% reduction) through systematic root cause analysis and sequential cleanup phases. Achieved zero warnings in production code and removed 22 unused dependencies for 15-25% expected compilation speedup. ## Phase Results ### Phase 1 (Agent 145): Critical Logic Bug Fixes - Fixed 18+ useless comparison warnings (logic errors) - Pattern: unsigned integers compared to zero (always true) - Files: 10 test files cleaned ### Phase 2 (Agent 146): Workspace-Wide Cargo Fix - Ran comprehensive cargo fix across all targets - 88 files modified (+202/-274 lines) - Warning reduction: 2484 → ~91 (96%) - Fixed 14 compilation errors introduced by cargo fix ### Phase 3 (Agent 147): Unused Dependency Removal - Removed 22 unused dependencies from 17 Cargo.toml files - Categories: tempfile (12), tracing-subscriber (8), proptest (3) - Expected speedup: 15-25% compilation time (~63 seconds saved) ### Phase 4a (Agent 148): Zero Warnings Achievement - Main workspace: 404 → 0 warnings (100% elimination) - Added Debug derives, prefixed unused variables - 16 files modified for final cleanup ### Phase 4b (Agent 149): CI Enforcement Validation - Verified existing RUSTFLAGS="-D warnings" in 5 workflows - Updated DEVELOPMENT.md documentation - Future warning accumulation: IMPOSSIBLE ✅ ## Files Modified (100+ total) Key Production Code: - trading_engine/src/types/circuit_breaker.rs: Debug derives - ml/src/safety/mod.rs: Unused variable fix - ml/src/integration/coordinator.rs: Unnecessary qualification fix - ml/src/integration/model_registry.rs: Conditional imports Critical Fixes: - trading_engine/src/lockfree/mod.rs: Restored pub use statements - risk/Cargo.toml: Added missing hdrhistogram dependency - tests/Cargo.toml: Added tracing-subscriber dependency - tli/src/tests.rs: Fixed logging initialization Load Tests: - services/load_tests/src/scenarios/*.rs: Cleaned up warnings - services/load_tests/src/metrics/metrics.rs: Added allow annotations 17 Cargo.toml files: Removed 22 unused dependencies ## Impact ✅ Production code: 0 warnings (100% clean) ✅ Test warnings: 2484 → 63 (97% reduction) ✅ Compilation speed: 15-25% faster (expected) ✅ Dependencies: 22 removed (cleaner graph) ✅ CI enforcement: Already active (future protection) ## Technical Insights **cargo fix Gotchas Discovered**: 1. Can remove critical pub use statements (false positive) 2. May remove imports still needed for tests 3. Doesn't validate dependency requirements → Always validate compilation after cargo fix **Warning Categories Fixed**: - Unused imports: ~50+ instances - Unused variables: ~30+ instances - Unused dependencies: 22 instances - Dead code: ~10+ instances - Logic bugs (useless comparisons): 18+ instances **Prevention**: CI enforces RUSTFLAGS="-D warnings" in 5 workflows 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
643 lines
22 KiB
Rust
643 lines
22 KiB
Rust
//! Database stress testing for PostgreSQL performance validation
|
|
//!
|
|
//! This test suite validates PostgreSQL can handle production load:
|
|
//! - 10,000 inserts/sec sustained for 60 seconds
|
|
//! - Concurrent writes (10, 100 connections)
|
|
//! - Connection pool behavior under stress
|
|
//! - Query performance degradation under load
|
|
//! - Transaction rollback performance
|
|
//!
|
|
//! Run with: cargo test -p load_tests --test database_stress_test -- --ignored --nocapture
|
|
|
|
use anyhow::Result;
|
|
use chrono::Utc;
|
|
use sqlx::postgres::{PgPool, PgPoolOptions};
|
|
use std::sync::atomic::{AtomicU64, Ordering};
|
|
use std::sync::Arc;
|
|
use std::time::{Duration, Instant};
|
|
use tokio::task::JoinSet;
|
|
use uuid::Uuid;
|
|
|
|
const DATABASE_URL: &str = "postgresql://foxhunt:foxhunt_dev_password@localhost:5432/foxhunt";
|
|
const TEST_SYMBOL: &str = "STRESS_TEST";
|
|
const TEST_ACCOUNT: &str = "stress_test_account";
|
|
|
|
/// Metrics for database operations
|
|
#[derive(Debug)]
|
|
struct DbMetrics {
|
|
inserts: AtomicU64,
|
|
selects: AtomicU64,
|
|
updates: AtomicU64,
|
|
errors: AtomicU64,
|
|
deadlocks: AtomicU64,
|
|
timeouts: AtomicU64,
|
|
}
|
|
|
|
impl DbMetrics {
|
|
fn new() -> Self {
|
|
Self {
|
|
inserts: AtomicU64::new(0),
|
|
selects: AtomicU64::new(0),
|
|
updates: AtomicU64::new(0),
|
|
errors: AtomicU64::new(0),
|
|
deadlocks: AtomicU64::new(0),
|
|
timeouts: AtomicU64::new(0),
|
|
}
|
|
}
|
|
|
|
fn print_summary(&self, duration: Duration, test_name: &str) {
|
|
let inserts = self.inserts.load(Ordering::Relaxed);
|
|
let selects = self.selects.load(Ordering::Relaxed);
|
|
let updates = self.updates.load(Ordering::Relaxed);
|
|
let errors = self.errors.load(Ordering::Relaxed);
|
|
let deadlocks = self.deadlocks.load(Ordering::Relaxed);
|
|
let timeouts = self.timeouts.load(Ordering::Relaxed);
|
|
|
|
let secs = duration.as_secs_f64();
|
|
let insert_rate = inserts as f64 / secs;
|
|
let select_rate = selects as f64 / secs;
|
|
let total_ops = inserts + selects + updates;
|
|
let error_rate = if total_ops > 0 {
|
|
(errors as f64 / total_ops as f64) * 100.0
|
|
} else {
|
|
0.0
|
|
};
|
|
|
|
println!("\n{}", "=".repeat(80));
|
|
println!("Database Stress Test: {}", test_name);
|
|
println!("{}", "=".repeat(80));
|
|
println!("Duration: {:.2}s", secs);
|
|
println!("Operations:");
|
|
println!(" Inserts: {} ({:.2}/sec)", inserts, insert_rate);
|
|
println!(" Selects: {} ({:.2}/sec)", selects, select_rate);
|
|
println!(" Updates: {}", updates);
|
|
println!("Errors:");
|
|
println!(" Total: {} ({:.2}%)", errors, error_rate);
|
|
println!(" Deadlocks: {}", deadlocks);
|
|
println!(" Timeouts: {}", timeouts);
|
|
println!("{}\n", "=".repeat(80));
|
|
}
|
|
}
|
|
|
|
/// Test 1: Baseline insert performance (single connection)
|
|
async fn test_baseline_insert_performance() -> Result<()> {
|
|
println!("\n🚀 Test 1: Baseline Insert Performance (single connection)");
|
|
|
|
let pool = PgPoolOptions::new()
|
|
.max_connections(1)
|
|
.connect(DATABASE_URL)
|
|
.await?;
|
|
|
|
let metrics = Arc::new(DbMetrics::new());
|
|
let start = Instant::now();
|
|
let test_duration = Duration::from_secs(10);
|
|
|
|
while start.elapsed() < test_duration {
|
|
let order_id = Uuid::new_v4();
|
|
let created_at = Utc::now().timestamp_nanos_opt().unwrap_or(0);
|
|
|
|
match sqlx::query(r#"
|
|
INSERT INTO orders (
|
|
id, symbol, side, order_type, time_in_force, quantity,
|
|
filled_quantity, remaining_quantity, status, created_at,
|
|
updated_at, account_id, venue
|
|
) VALUES ($1, $2, 'buy', 'market', 'day', 100, 0, 100, 'pending', $3, $3, $4, 'test')
|
|
"#)
|
|
.bind(order_id)
|
|
.bind(TEST_SYMBOL)
|
|
.bind(created_at)
|
|
.bind(TEST_ACCOUNT)
|
|
.execute(&pool)
|
|
.await
|
|
{
|
|
Ok(_) => metrics.inserts.fetch_add(1, Ordering::Relaxed),
|
|
Err(e) => {
|
|
eprintln!("Insert error: {:?}", e);
|
|
metrics.errors.fetch_add(1, Ordering::Relaxed)
|
|
}
|
|
};
|
|
}
|
|
|
|
let duration = start.elapsed();
|
|
metrics.print_summary(duration, "Baseline Insert Performance");
|
|
|
|
let insert_rate = metrics.inserts.load(Ordering::Relaxed) as f64 / duration.as_secs_f64();
|
|
println!("✅ Baseline: {:.2} inserts/sec", insert_rate);
|
|
|
|
// Cleanup
|
|
cleanup_test_data(&pool).await?;
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Test 2: Concurrent writes (10 connections)
|
|
async fn test_concurrent_writes_10_connections() -> Result<()> {
|
|
println!("\n🚀 Test 2: Concurrent Writes (10 connections)");
|
|
|
|
let pool = PgPoolOptions::new()
|
|
.max_connections(10)
|
|
.connect(DATABASE_URL)
|
|
.await?;
|
|
|
|
let metrics = Arc::new(DbMetrics::new());
|
|
let mut join_set = JoinSet::new();
|
|
let test_duration = Duration::from_secs(30);
|
|
|
|
for worker_id in 0..10 {
|
|
let pool = pool.clone();
|
|
let metrics = Arc::clone(&metrics);
|
|
|
|
join_set.spawn(async move {
|
|
let start = Instant::now();
|
|
while start.elapsed() < test_duration {
|
|
let order_id = Uuid::new_v4();
|
|
let created_at = Utc::now().timestamp_nanos_opt().unwrap_or(0);
|
|
let symbol = format!("{}_W{}", TEST_SYMBOL, worker_id);
|
|
|
|
match sqlx::query(r#"
|
|
INSERT INTO orders (
|
|
id, symbol, side, order_type, time_in_force, quantity,
|
|
filled_quantity, remaining_quantity, status, created_at,
|
|
updated_at, account_id, venue
|
|
) VALUES ($1, $2, 'buy', 'market', 'day', 100, 0, 100, 'pending', $3, $3, $4, 'test')
|
|
"#)
|
|
.bind(order_id)
|
|
.bind(symbol)
|
|
.bind(created_at)
|
|
.bind(TEST_ACCOUNT)
|
|
.execute(&pool)
|
|
.await
|
|
{
|
|
Ok(_) => metrics.inserts.fetch_add(1, Ordering::Relaxed),
|
|
Err(e) => {
|
|
if e.to_string().contains("deadlock") {
|
|
metrics.deadlocks.fetch_add(1, Ordering::Relaxed);
|
|
}
|
|
metrics.errors.fetch_add(1, Ordering::Relaxed)
|
|
}
|
|
};
|
|
|
|
tokio::time::sleep(Duration::from_micros(1000)).await;
|
|
}
|
|
});
|
|
}
|
|
|
|
let start = Instant::now();
|
|
while let Some(result) = join_set.join_next().await {
|
|
if let Err(e) = result {
|
|
eprintln!("Worker error: {:?}", e);
|
|
}
|
|
}
|
|
let duration = start.elapsed();
|
|
|
|
metrics.print_summary(duration, "Concurrent Writes (10 connections)");
|
|
|
|
let insert_rate = metrics.inserts.load(Ordering::Relaxed) as f64 / duration.as_secs_f64();
|
|
println!("✅ Throughput: {:.2} inserts/sec", insert_rate);
|
|
println!("✅ Deadlocks: {}", metrics.deadlocks.load(Ordering::Relaxed));
|
|
|
|
// Cleanup
|
|
cleanup_test_data(&pool).await?;
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Test 3: High throughput (100 connections, target 10K inserts/sec)
|
|
async fn test_high_throughput_100_connections() -> Result<()> {
|
|
println!("\n🚀 Test 3: High Throughput (100 connections, 60s sustained)");
|
|
|
|
let pool = PgPoolOptions::new()
|
|
.max_connections(100)
|
|
.acquire_timeout(Duration::from_secs(5))
|
|
.connect(DATABASE_URL)
|
|
.await?;
|
|
|
|
let metrics = Arc::new(DbMetrics::new());
|
|
let mut join_set = JoinSet::new();
|
|
let test_duration = Duration::from_secs(60);
|
|
|
|
for worker_id in 0..100 {
|
|
let pool = pool.clone();
|
|
let metrics = Arc::clone(&metrics);
|
|
|
|
join_set.spawn(async move {
|
|
let start = Instant::now();
|
|
while start.elapsed() < test_duration {
|
|
let order_id = Uuid::new_v4();
|
|
let created_at = Utc::now().timestamp_nanos_opt().unwrap_or(0);
|
|
let symbol = format!("{}_W{}", TEST_SYMBOL, worker_id % 20);
|
|
|
|
match sqlx::query(r#"
|
|
INSERT INTO orders (
|
|
id, symbol, side, order_type, time_in_force, quantity,
|
|
filled_quantity, remaining_quantity, status, created_at,
|
|
updated_at, account_id, venue
|
|
) VALUES ($1, $2, 'buy', 'market', 'day', 100, 0, 100, 'pending', $3, $3, $4, 'test')
|
|
"#)
|
|
.bind(order_id)
|
|
.bind(symbol)
|
|
.bind(created_at)
|
|
.bind(TEST_ACCOUNT)
|
|
.execute(&pool)
|
|
.await
|
|
{
|
|
Ok(_) => metrics.inserts.fetch_add(1, Ordering::Relaxed),
|
|
Err(e) => {
|
|
let err_str = e.to_string();
|
|
if err_str.contains("deadlock") {
|
|
metrics.deadlocks.fetch_add(1, Ordering::Relaxed);
|
|
} else if err_str.contains("timeout") || err_str.contains("timed out") {
|
|
metrics.timeouts.fetch_add(1, Ordering::Relaxed);
|
|
}
|
|
metrics.errors.fetch_add(1, Ordering::Relaxed)
|
|
}
|
|
};
|
|
|
|
// Target: 100 inserts/sec per worker = 10K total
|
|
tokio::time::sleep(Duration::from_micros(10000)).await;
|
|
}
|
|
});
|
|
}
|
|
|
|
let start = Instant::now();
|
|
while let Some(result) = join_set.join_next().await {
|
|
if let Err(e) = result {
|
|
eprintln!("Worker error: {:?}", e);
|
|
}
|
|
}
|
|
let duration = start.elapsed();
|
|
|
|
metrics.print_summary(duration, "High Throughput (100 connections)");
|
|
|
|
let insert_rate = metrics.inserts.load(Ordering::Relaxed) as f64 / duration.as_secs_f64();
|
|
let inserts = metrics.inserts.load(Ordering::Relaxed);
|
|
let errors = metrics.errors.load(Ordering::Relaxed);
|
|
|
|
println!("🎯 Target: 10,000 inserts/sec");
|
|
println!("✅ Achieved: {:.2} inserts/sec", insert_rate);
|
|
println!("✅ Success rate: {:.2}%", (inserts as f64 / (inserts + errors) as f64) * 100.0);
|
|
|
|
assert!(
|
|
insert_rate >= 9000.0,
|
|
"Insert rate too low: {:.2} (expected >= 9000)",
|
|
insert_rate
|
|
);
|
|
|
|
// Cleanup
|
|
cleanup_test_data(&pool).await?;
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Test 4: Connection pool stress (exceed pool limits)
|
|
async fn test_connection_pool_stress() -> Result<()> {
|
|
println!("\n🚀 Test 4: Connection Pool Stress (150 tasks, 100 max connections)");
|
|
|
|
let pool = PgPoolOptions::new()
|
|
.max_connections(100)
|
|
.acquire_timeout(Duration::from_secs(10))
|
|
.connect(DATABASE_URL)
|
|
.await?;
|
|
|
|
let metrics = Arc::new(DbMetrics::new());
|
|
let mut join_set = JoinSet::new();
|
|
let test_duration = Duration::from_secs(20);
|
|
|
|
// Spawn 150 tasks (more than pool size)
|
|
for worker_id in 0..150 {
|
|
let pool = pool.clone();
|
|
let metrics = Arc::clone(&metrics);
|
|
|
|
join_set.spawn(async move {
|
|
let start = Instant::now();
|
|
let mut wait_times = Vec::new();
|
|
|
|
while start.elapsed() < test_duration {
|
|
let acquire_start = Instant::now();
|
|
let order_id = Uuid::new_v4();
|
|
let created_at = Utc::now().timestamp_nanos_opt().unwrap_or(0);
|
|
|
|
match sqlx::query(r#"
|
|
INSERT INTO orders (
|
|
id, symbol, side, order_type, time_in_force, quantity,
|
|
filled_quantity, remaining_quantity, status, created_at,
|
|
updated_at, account_id, venue
|
|
) VALUES ($1, $2, 'buy', 'market', 'day', 100, 0, 100, 'pending', $3, $3, $4, 'test')
|
|
"#)
|
|
.bind(order_id)
|
|
.bind(format!("{}_P{}", TEST_SYMBOL, worker_id % 30))
|
|
.bind(created_at)
|
|
.bind(TEST_ACCOUNT)
|
|
.execute(&pool)
|
|
.await
|
|
{
|
|
Ok(_) => {
|
|
metrics.inserts.fetch_add(1, Ordering::Relaxed);
|
|
wait_times.push(acquire_start.elapsed().as_millis());
|
|
}
|
|
Err(e) => {
|
|
if e.to_string().contains("timeout") {
|
|
metrics.timeouts.fetch_add(1, Ordering::Relaxed);
|
|
}
|
|
metrics.errors.fetch_add(1, Ordering::Relaxed);
|
|
}
|
|
};
|
|
|
|
tokio::time::sleep(Duration::from_millis(20)).await;
|
|
}
|
|
|
|
if !wait_times.is_empty() {
|
|
let avg_wait = wait_times.iter().sum::<u128>() / wait_times.len() as u128;
|
|
let max_wait = wait_times.iter().max().unwrap_or(&0);
|
|
println!(
|
|
"Worker {}: avg wait {}ms, max wait {}ms",
|
|
worker_id, avg_wait, max_wait
|
|
);
|
|
}
|
|
});
|
|
}
|
|
|
|
let start = Instant::now();
|
|
while let Some(result) = join_set.join_next().await {
|
|
if let Err(e) = result {
|
|
eprintln!("Worker error: {:?}", e);
|
|
}
|
|
}
|
|
let duration = start.elapsed();
|
|
|
|
metrics.print_summary(duration, "Connection Pool Stress");
|
|
|
|
let timeouts = metrics.timeouts.load(Ordering::Relaxed);
|
|
let errors = metrics.errors.load(Ordering::Relaxed);
|
|
|
|
println!("✅ Connection timeouts: {}", timeouts);
|
|
println!("✅ Total errors: {}", errors);
|
|
|
|
assert!(
|
|
errors < 100,
|
|
"Too many errors under pool stress: {} (expected < 100)",
|
|
errors
|
|
);
|
|
|
|
// Cleanup
|
|
cleanup_test_data(&pool).await?;
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Test 5: Query performance under write load
|
|
async fn test_query_performance_under_load() -> Result<()> {
|
|
println!("\n🚀 Test 5: Query Performance Under Write Load");
|
|
|
|
let pool = PgPoolOptions::new()
|
|
.max_connections(50)
|
|
.connect(DATABASE_URL)
|
|
.await?;
|
|
|
|
let metrics = Arc::new(DbMetrics::new());
|
|
let mut join_set = JoinSet::new();
|
|
let test_duration = Duration::from_secs(30);
|
|
|
|
// Spawn 30 writers
|
|
for writer_id in 0..30 {
|
|
let pool = pool.clone();
|
|
let metrics = Arc::clone(&metrics);
|
|
|
|
join_set.spawn(async move {
|
|
let start = Instant::now();
|
|
while start.elapsed() < test_duration {
|
|
let order_id = Uuid::new_v4();
|
|
let created_at = Utc::now().timestamp_nanos_opt().unwrap_or(0);
|
|
|
|
if sqlx::query(r#"
|
|
INSERT INTO orders (
|
|
id, symbol, side, order_type, time_in_force, quantity,
|
|
filled_quantity, remaining_quantity, status, created_at,
|
|
updated_at, account_id, venue
|
|
) VALUES ($1, $2, 'buy', 'market', 'day', 100, 0, 100, 'pending', $3, $3, $4, 'test')
|
|
"#)
|
|
.bind(order_id)
|
|
.bind(format!("{}_Q{}", TEST_SYMBOL, writer_id))
|
|
.bind(created_at)
|
|
.bind(TEST_ACCOUNT)
|
|
.execute(&pool)
|
|
.await
|
|
.is_ok()
|
|
{
|
|
metrics.inserts.fetch_add(1, Ordering::Relaxed);
|
|
}
|
|
|
|
tokio::time::sleep(Duration::from_millis(10)).await;
|
|
}
|
|
});
|
|
}
|
|
|
|
// Spawn 20 readers
|
|
for reader_id in 0..20 {
|
|
let pool = pool.clone();
|
|
let metrics = Arc::clone(&metrics);
|
|
|
|
join_set.spawn(async move {
|
|
let start = Instant::now();
|
|
let mut query_times = Vec::new();
|
|
|
|
while start.elapsed() < test_duration {
|
|
let query_start = Instant::now();
|
|
|
|
match sqlx::query(r#"
|
|
SELECT id, symbol, status, quantity, filled_quantity
|
|
FROM orders
|
|
WHERE account_id = $1 AND status = 'pending'
|
|
ORDER BY created_at DESC
|
|
LIMIT 100
|
|
"#)
|
|
.bind(TEST_ACCOUNT)
|
|
.fetch_all(&pool)
|
|
.await
|
|
{
|
|
Ok(_) => {
|
|
metrics.selects.fetch_add(1, Ordering::Relaxed);
|
|
query_times.push(query_start.elapsed().as_micros());
|
|
}
|
|
Err(_) => {
|
|
metrics.errors.fetch_add(1, Ordering::Relaxed);
|
|
}
|
|
}
|
|
|
|
tokio::time::sleep(Duration::from_millis(50)).await;
|
|
}
|
|
|
|
if !query_times.is_empty() {
|
|
let avg_time = query_times.iter().sum::<u128>() / query_times.len() as u128;
|
|
let p95_idx = (query_times.len() as f64 * 0.95) as usize;
|
|
let mut sorted = query_times.clone();
|
|
sorted.sort_unstable();
|
|
let p95_time = sorted.get(p95_idx).unwrap_or(&0);
|
|
|
|
println!(
|
|
"Reader {}: avg {}μs, p95 {}μs",
|
|
reader_id, avg_time, p95_time
|
|
);
|
|
}
|
|
});
|
|
}
|
|
|
|
let start = Instant::now();
|
|
while let Some(result) = join_set.join_next().await {
|
|
if let Err(e) = result {
|
|
eprintln!("Worker error: {:?}", e);
|
|
}
|
|
}
|
|
let duration = start.elapsed();
|
|
|
|
metrics.print_summary(duration, "Query Performance Under Load");
|
|
|
|
let selects = metrics.selects.load(Ordering::Relaxed);
|
|
let select_rate = selects as f64 / duration.as_secs_f64();
|
|
|
|
println!("✅ Read throughput: {:.2} queries/sec", select_rate);
|
|
|
|
// Cleanup
|
|
cleanup_test_data(&pool).await?;
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Test 6: Transaction stress (with rollbacks)
|
|
async fn test_transaction_stress() -> Result<()> {
|
|
println!("\n🚀 Test 6: Transaction Stress (commits and rollbacks)");
|
|
|
|
let pool = PgPoolOptions::new()
|
|
.max_connections(20)
|
|
.connect(DATABASE_URL)
|
|
.await?;
|
|
|
|
let metrics = Arc::new(DbMetrics::new());
|
|
let mut join_set = JoinSet::new();
|
|
let test_duration = Duration::from_secs(20);
|
|
|
|
for worker_id in 0..20 {
|
|
let pool = pool.clone();
|
|
let metrics = Arc::clone(&metrics);
|
|
|
|
join_set.spawn(async move {
|
|
let start = Instant::now();
|
|
let mut commits = 0u64;
|
|
let mut rollbacks = 0u64;
|
|
|
|
while start.elapsed() < test_duration {
|
|
let mut tx = match pool.begin().await {
|
|
Ok(tx) => tx,
|
|
Err(_) => {
|
|
metrics.errors.fetch_add(1, Ordering::Relaxed);
|
|
continue;
|
|
}
|
|
};
|
|
|
|
let order_id = Uuid::new_v4();
|
|
let created_at = Utc::now().timestamp_nanos_opt().unwrap_or(0);
|
|
|
|
if sqlx::query(r#"
|
|
INSERT INTO orders (
|
|
id, symbol, side, order_type, time_in_force, quantity,
|
|
filled_quantity, remaining_quantity, status, created_at,
|
|
updated_at, account_id, venue
|
|
) VALUES ($1, $2, 'buy', 'market', 'day', 100, 0, 100, 'pending', $3, $3, $4, 'test')
|
|
"#)
|
|
.bind(order_id)
|
|
.bind(format!("{}_T{}", TEST_SYMBOL, worker_id))
|
|
.bind(created_at)
|
|
.bind(TEST_ACCOUNT)
|
|
.execute(&mut *tx)
|
|
.await
|
|
.is_ok()
|
|
{
|
|
metrics.inserts.fetch_add(1, Ordering::Relaxed);
|
|
|
|
// Randomly commit or rollback (70% commit, 30% rollback)
|
|
if worker_id % 10 < 7 {
|
|
if tx.commit().await.is_ok() {
|
|
commits += 1;
|
|
}
|
|
} else {
|
|
if tx.rollback().await.is_ok() {
|
|
rollbacks += 1;
|
|
}
|
|
}
|
|
}
|
|
|
|
tokio::time::sleep(Duration::from_millis(10)).await;
|
|
}
|
|
|
|
println!(
|
|
"Worker {}: commits={}, rollbacks={}",
|
|
worker_id, commits, rollbacks
|
|
);
|
|
});
|
|
}
|
|
|
|
let start = Instant::now();
|
|
while let Some(result) = join_set.join_next().await {
|
|
if let Err(e) = result {
|
|
eprintln!("Worker error: {:?}", e);
|
|
}
|
|
}
|
|
let duration = start.elapsed();
|
|
|
|
metrics.print_summary(duration, "Transaction Stress");
|
|
|
|
let tx_rate = metrics.inserts.load(Ordering::Relaxed) as f64 / duration.as_secs_f64();
|
|
println!("✅ Transaction rate: {:.2} tx/sec", tx_rate);
|
|
|
|
// Cleanup
|
|
cleanup_test_data(&pool).await?;
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Cleanup test data
|
|
async fn cleanup_test_data(pool: &PgPool) -> Result<()> {
|
|
println!("🧹 Cleaning up test data...");
|
|
|
|
let result = sqlx::query(r#"DELETE FROM orders WHERE symbol LIKE $1"#)
|
|
.bind(format!("{}%", TEST_SYMBOL))
|
|
.execute(pool)
|
|
.await?;
|
|
|
|
println!("🧹 Deleted {} test orders", result.rows_affected());
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Integration test: Run all database stress tests
|
|
#[tokio::test]
|
|
#[ignore]
|
|
async fn test_comprehensive_database_stress() -> Result<()> {
|
|
println!("\n{}", "=".repeat(80));
|
|
println!("🎯 Comprehensive Database Stress Test Suite");
|
|
println!("{}\n", "=".repeat(80));
|
|
|
|
test_baseline_insert_performance().await?;
|
|
tokio::time::sleep(Duration::from_secs(2)).await;
|
|
|
|
test_concurrent_writes_10_connections().await?;
|
|
tokio::time::sleep(Duration::from_secs(2)).await;
|
|
|
|
test_high_throughput_100_connections().await?;
|
|
tokio::time::sleep(Duration::from_secs(2)).await;
|
|
|
|
test_connection_pool_stress().await?;
|
|
tokio::time::sleep(Duration::from_secs(2)).await;
|
|
|
|
test_query_performance_under_load().await?;
|
|
tokio::time::sleep(Duration::from_secs(2)).await;
|
|
|
|
test_transaction_stress().await?;
|
|
|
|
println!("\n{}", "=".repeat(80));
|
|
println!("✅ All database stress tests completed successfully!");
|
|
println!("{}\n", "=".repeat(80));
|
|
|
|
Ok(())
|
|
}
|