use crate::{
ConformanceTest, MpscReceiver, MpscSender, OneshotReceiver, OneshotSender, RuntimeInterface,
TestCategory, TestMeta, TestResult, checkpoint,
};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::time::{Duration, Instant};
pub fn all_tests<RT: RuntimeInterface>() -> Vec<ConformanceTest<RT>> {
vec![
sc_001_context_scope_ownership::<RT>(),
sc_002_context_cannot_escape_scope::<RT>(),
sc_003_request_drain_finalize_lifecycle::<RT>(),
sc_004_region_close_implies_quiescence::<RT>(),
sc_005_no_obligation_leaks::<RT>(),
sc_006_nested_scope_ownership::<RT>(),
sc_007_cancel_propagation_hierarchy::<RT>(),
sc_008_resource_cleanup_on_scope_exit::<RT>(),
]
}
pub fn sc_001_context_scope_ownership<RT: RuntimeInterface>() -> ConformanceTest<RT> {
ConformanceTest::new(
TestMeta {
id: "sc-001".to_string(),
name: "Context scope ownership - lifetime management".to_string(),
description: "Context must properly manage lifetime relative to request scope"
.to_string(),
category: TestCategory::Spawn,
tags: vec![
"structured".to_string(),
"scope".to_string(),
"context".to_string(),
],
expected: "Context outlives request scope without leaks".to_string(),
},
|rt| {
rt.block_on(async {
checkpoint(
"Starting context scope ownership test",
serde_json::json!({}),
);
let completion_flag = Arc::new(AtomicBool::new(false));
let completion_flag_clone = completion_flag.clone();
let task = rt.spawn(async move {
checkpoint("Creating scoped context", serde_json::json!({}));
let context_active = Arc::new(AtomicBool::new(true));
let context_active_clone = context_active.clone();
let _background_task = rt.spawn(async move {
while context_active_clone.load(Ordering::Acquire) {
rt.sleep(Duration::from_millis(1)).await;
}
checkpoint("Background work completed", serde_json::json!({}));
});
rt.sleep(Duration::from_millis(10)).await;
if !context_active.load(Ordering::Acquire) {
return Err("Context should remain active during request");
}
context_active.store(false, Ordering::Release);
completion_flag_clone.store(true, Ordering::Release);
checkpoint("Request scope ended", serde_json::json!({}));
Ok(())
});
let result = rt.timeout(Duration::from_secs(1), task).await;
match result {
Ok(Ok(())) => {
if completion_flag.load(Ordering::Acquire) {
checkpoint(
"Context scope ownership test passed",
serde_json::json!({}),
);
TestResult::passed()
} else {
TestResult::failed("Task completed but flag not set")
}
}
Ok(Err(e)) => {
checkpoint("Task failed", serde_json::json!({"error": e}));
TestResult::failed("Task execution failed")
}
Err(_) => {
checkpoint("Test timed out", serde_json::json!({}));
TestResult::failed("Test timed out - possible deadlock")
}
}
})
},
)
}
pub fn sc_002_context_cannot_escape_scope<RT: RuntimeInterface>() -> ConformanceTest<RT> {
ConformanceTest::new(
TestMeta {
id: "sc-002".to_string(),
name: "Context cannot escape scope".to_string(),
description: "Context must not be accessible outside its defining scope".to_string(),
category: TestCategory::Spawn,
tags: vec![
"structured".to_string(),
"scope".to_string(),
"context".to_string(),
],
expected: "Context cannot outlive scope boundaries".to_string(),
},
|rt| {
rt.block_on(async {
checkpoint(
"Starting context escape prevention test",
serde_json::json!({}),
);
let (tx, rx) = rt.oneshot();
let task = rt.spawn(async move {
let context_id = {
let context_id = Arc::new(AtomicU64::new(42));
context_id.load(Ordering::Acquire)
};
tx.send(context_id).expect("Send should succeed");
Ok(())
});
let result = rt.timeout(Duration::from_secs(1), task).await;
match result {
Ok(Ok(())) => {
let received = rx.await.expect("Should receive value");
if received == 42 {
checkpoint(
"Context escape prevention test passed",
serde_json::json!({}),
);
TestResult::passed()
} else {
TestResult::failed("Context value not preserved within scope")
}
}
_ => TestResult::failed("Context scope test failed"),
}
})
},
)
}
pub fn sc_003_request_drain_finalize_lifecycle<RT: RuntimeInterface>() -> ConformanceTest<RT> {
ConformanceTest::new(
TestMeta {
id: "sc-003".to_string(),
name: "Request-drain-finalize lifecycle".to_string(),
description: "Async operations must follow proper request→drain→finalize phases"
.to_string(),
category: TestCategory::Spawn,
tags: vec![
"structured".to_string(),
"lifecycle".to_string(),
"phases".to_string(),
],
expected: "Operations follow request→drain→finalize sequence".to_string(),
},
|rt| {
rt.block_on(async {
checkpoint(
"Starting request-drain-finalize lifecycle test",
serde_json::json!({}),
);
let phase_counter = Arc::new(AtomicUsize::new(0));
let phase_counter_clone = phase_counter.clone();
let task = rt.spawn(async move {
checkpoint("Phase 1: Request", serde_json::json!({}));
phase_counter_clone.store(1, Ordering::Release);
rt.sleep(Duration::from_millis(5)).await;
checkpoint("Phase 2: Drain", serde_json::json!({}));
phase_counter_clone.store(2, Ordering::Release);
rt.sleep(Duration::from_millis(5)).await;
checkpoint("Phase 3: Finalize", serde_json::json!({}));
phase_counter_clone.store(3, Ordering::Release);
Ok(())
});
let result = rt.timeout(Duration::from_secs(1), task).await;
match result {
Ok(Ok(())) => {
let final_phase = phase_counter.load(Ordering::Acquire);
if final_phase == 3 {
checkpoint(
"Request-drain-finalize lifecycle test passed",
serde_json::json!({"final_phase": final_phase}),
);
TestResult::passed()
} else {
TestResult::failed(format!(
"Lifecycle incomplete, stopped at phase {}",
final_phase
))
}
}
_ => TestResult::failed("Lifecycle test failed"),
}
})
},
)
}
pub fn sc_004_region_close_implies_quiescence<RT: RuntimeInterface>() -> ConformanceTest<RT> {
ConformanceTest::new(
TestMeta {
id: "sc-004".to_string(),
name: "Region close implies quiescence".to_string(),
description: "Region closure must result in quiescent state".to_string(),
category: TestCategory::Spawn,
tags: vec![
"structured".to_string(),
"region".to_string(),
"quiescence".to_string(),
],
expected: "Region close results in quiescent state".to_string(),
},
|rt| {
rt.block_on(async {
checkpoint(
"Starting region close quiescence test",
serde_json::json!({}),
);
let active_tasks = Arc::new(AtomicUsize::new(0));
let region_closed = Arc::new(AtomicBool::new(false));
let tasks = {
let mut tasks = Vec::new();
let active_tasks_clone = active_tasks.clone();
let region_closed_clone = region_closed.clone();
for i in 0..3 {
let active_tasks = active_tasks_clone.clone();
let region_closed = region_closed_clone.clone();
let task = rt.spawn(async move {
active_tasks.fetch_add(1, Ordering::SeqCst);
checkpoint("Task started", serde_json::json!({"task_id": i}));
while !region_closed.load(Ordering::Acquire) {
rt.sleep(Duration::from_millis(1)).await;
}
checkpoint("Task shutting down", serde_json::json!({"task_id": i}));
active_tasks.fetch_sub(1, Ordering::SeqCst);
Ok(())
});
tasks.push(task);
}
tasks
};
while active_tasks.load(Ordering::Acquire) < 3 {
rt.sleep(Duration::from_millis(1)).await;
}
checkpoint("Closing region", serde_json::json!({}));
region_closed.store(true, Ordering::Release);
let quiescence_start = Instant::now();
let mut all_completed = true;
for task in tasks {
let result = rt.timeout(Duration::from_millis(500), task).await;
if result.is_err() {
all_completed = false;
break;
}
}
let final_active = active_tasks.load(Ordering::Acquire);
if all_completed && final_active == 0 {
let quiescence_duration = quiescence_start.elapsed();
checkpoint(
"Quiescence achieved",
serde_json::json!({"duration_ms": quiescence_duration.as_millis()}),
);
TestResult::passed()
} else {
TestResult::failed(format!(
"Quiescence not achieved: {} tasks still active, completion: {}",
final_active, all_completed
))
}
})
},
)
}
pub fn sc_005_no_obligation_leaks<RT: RuntimeInterface>() -> ConformanceTest<RT> {
ConformanceTest::new(
TestMeta {
id: "sc-005".to_string(),
name: "No obligation leaks after region close".to_string(),
description: "Region close must not leak obligations".to_string(),
category: TestCategory::Spawn,
tags: vec![
"structured".to_string(),
"obligations".to_string(),
"leak".to_string(),
],
expected: "No obligation leaks after region close".to_string(),
},
|rt| {
rt.block_on(async {
checkpoint(
"Starting obligation leak prevention test",
serde_json::json!({}),
);
let obligation_count = Arc::new(AtomicUsize::new(0));
let obligation_count_clone = obligation_count.clone();
let region_task = rt.spawn(async move {
for i in 0..5 {
obligation_count_clone.fetch_add(1, Ordering::SeqCst);
checkpoint(
"Created obligation",
serde_json::json!({"obligation_id": i}),
);
rt.sleep(Duration::from_millis(2)).await;
}
rt.sleep(Duration::from_millis(10)).await;
while obligation_count_clone.load(Ordering::Acquire) > 0 {
let current = obligation_count_clone.fetch_sub(1, Ordering::SeqCst);
if current > 0 {
checkpoint(
"Cleaned obligation",
serde_json::json!({"obligation_id": current - 1}),
);
}
rt.sleep(Duration::from_millis(1)).await;
}
checkpoint("All obligations cleaned up", serde_json::json!({}));
Ok(())
});
let result = rt.timeout(Duration::from_secs(1), region_task).await;
match result {
Ok(Ok(())) => {
let final_obligations = obligation_count.load(Ordering::Acquire);
if final_obligations == 0 {
checkpoint("No obligation leaks detected", serde_json::json!({}));
TestResult::passed()
} else {
TestResult::failed(format!("Leaked {} obligations", final_obligations))
}
}
_ => TestResult::failed("Obligation leak test failed"),
}
})
},
)
}
pub fn sc_006_nested_scope_ownership<RT: RuntimeInterface>() -> ConformanceTest<RT> {
ConformanceTest::new(
TestMeta {
id: "sc-006".to_string(),
name: "Nested scope ownership laws".to_string(),
description: "Nested scopes must maintain proper ownership hierarchy".to_string(),
category: TestCategory::Spawn,
tags: vec![
"structured".to_string(),
"nested".to_string(),
"ownership".to_string(),
],
expected: "Nested scopes maintain proper hierarchy".to_string(),
},
|rt| {
rt.block_on(async {
checkpoint(
"Starting nested scope ownership test",
serde_json::json!({}),
);
let outer_scope_active = Arc::new(AtomicBool::new(true));
let inner_scope_active = Arc::new(AtomicBool::new(false));
let test_completed = Arc::new(AtomicBool::new(false));
let outer_scope_active_clone = outer_scope_active.clone();
let inner_scope_active_clone = inner_scope_active.clone();
let test_completed_clone = test_completed.clone();
let task = rt.spawn(async move {
checkpoint("Entering outer scope", serde_json::json!({}));
{
if !outer_scope_active_clone.load(Ordering::Acquire) {
return Err("Outer scope should be active");
}
{
checkpoint("Entering inner scope", serde_json::json!({}));
inner_scope_active_clone.store(true, Ordering::Release);
if !outer_scope_active_clone.load(Ordering::Acquire)
|| !inner_scope_active_clone.load(Ordering::Acquire)
{
return Err("Both scopes should be active");
}
rt.sleep(Duration::from_millis(5)).await;
checkpoint("Exiting inner scope", serde_json::json!({}));
inner_scope_active_clone.store(false, Ordering::Release);
}
if !outer_scope_active_clone.load(Ordering::Acquire)
|| inner_scope_active_clone.load(Ordering::Acquire)
{
return Err("Incorrect scope state after inner scope exit");
}
rt.sleep(Duration::from_millis(5)).await;
checkpoint("Exiting outer scope", serde_json::json!({}));
outer_scope_active_clone.store(false, Ordering::Release);
}
if outer_scope_active_clone.load(Ordering::Acquire)
|| inner_scope_active_clone.load(Ordering::Acquire)
{
return Err("Scopes should be inactive after exit");
}
test_completed_clone.store(true, Ordering::Release);
checkpoint("Nested scope test completed", serde_json::json!({}));
Ok(())
});
let result = rt.timeout(Duration::from_secs(1), task).await;
match result {
Ok(Ok(())) => {
if test_completed.load(Ordering::Acquire) {
TestResult::passed()
} else {
TestResult::failed("Test did not complete properly")
}
}
Ok(Err(e)) => {
TestResult::failed(format!("Nested scope ownership test failed: {}", e))
}
Err(_) => TestResult::failed("Test timed out"),
}
})
},
)
}
pub fn sc_007_cancel_propagation_hierarchy<RT: RuntimeInterface>() -> ConformanceTest<RT> {
ConformanceTest::new(
TestMeta {
id: "sc-007".to_string(),
name: "Cancel propagation hierarchy".to_string(),
description: "Cancellation must propagate correctly through scope hierarchy"
.to_string(),
category: TestCategory::Cancel,
tags: vec![
"structured".to_string(),
"cancel".to_string(),
"hierarchy".to_string(),
],
expected: "Cancellation propagates through scope hierarchy".to_string(),
},
|rt| {
rt.block_on(async {
checkpoint(
"Starting cancel propagation hierarchy test",
serde_json::json!({}),
);
let cancel_signal = Arc::new(AtomicBool::new(false));
let child_cancelled = Arc::new(AtomicBool::new(false));
let cancel_signal_clone = cancel_signal.clone();
let child_cancelled_clone = child_cancelled.clone();
let parent_task = rt.spawn(async move {
checkpoint("Parent task started", serde_json::json!({}));
let child_task = rt.spawn(async move {
checkpoint("Child task started", serde_json::json!({}));
while !cancel_signal_clone.load(Ordering::Acquire) {
rt.sleep(Duration::from_millis(1)).await;
}
checkpoint("Child task received cancellation", serde_json::json!({}));
child_cancelled_clone.store(true, Ordering::Release);
Ok(())
});
rt.sleep(Duration::from_millis(10)).await;
checkpoint("Parent signaling cancellation", serde_json::json!({}));
cancel_signal_clone.store(true, Ordering::Release);
let child_result = rt.timeout(Duration::from_millis(100), child_task).await;
match child_result {
Ok(Ok(())) => {
checkpoint("Child task completed gracefully", serde_json::json!({}));
Ok(())
}
_ => {
checkpoint(
"Child task did not complete properly",
serde_json::json!({}),
);
Err("Child cancellation failed")
}
}
});
let result = rt.timeout(Duration::from_secs(1), parent_task).await;
match result {
Ok(Ok(())) => {
if child_cancelled.load(Ordering::Acquire) {
checkpoint(
"Cancel propagation hierarchy test passed",
serde_json::json!({}),
);
TestResult::passed()
} else {
TestResult::failed("Child task was not cancelled")
}
}
Ok(Err(e)) => {
TestResult::failed(format!("Cancel propagation test failed: {}", e))
}
Err(_) => TestResult::failed("Test timed out"),
}
})
},
)
}
pub fn sc_008_resource_cleanup_on_scope_exit<RT: RuntimeInterface>() -> ConformanceTest<RT> {
ConformanceTest::new(
TestMeta {
id: "sc-008".to_string(),
name: "Resource cleanup on scope exit".to_string(),
description: "Resources must be cleaned up when scopes exit".to_string(),
category: TestCategory::Spawn,
tags: vec![
"structured".to_string(),
"cleanup".to_string(),
"resources".to_string(),
],
expected: "Resources cleaned up on scope exit".to_string(),
},
|rt| {
rt.block_on(async {
checkpoint("Starting resource cleanup test", serde_json::json!({}));
let resource_count = Arc::new(AtomicUsize::new(0));
let cleanup_called = Arc::new(AtomicBool::new(false));
let resource_count_clone = resource_count.clone();
let cleanup_called_clone = cleanup_called.clone();
let task = rt.spawn(async move {
{
checkpoint("Acquiring resources", serde_json::json!({}));
for i in 0..3 {
resource_count_clone.fetch_add(1, Ordering::SeqCst);
checkpoint("Acquired resource", serde_json::json!({"resource_id": i}));
}
rt.sleep(Duration::from_millis(10)).await;
checkpoint(
"Scope ending - cleaning up resources",
serde_json::json!({}),
);
while resource_count_clone.load(Ordering::Acquire) > 0 {
let current = resource_count_clone.fetch_sub(1, Ordering::SeqCst);
if current > 0 {
checkpoint(
"Released resource",
serde_json::json!({"resource_id": current - 1}),
);
}
}
cleanup_called_clone.store(true, Ordering::Release);
}
checkpoint("Scope exited", serde_json::json!({}));
Ok(())
});
let result = rt.timeout(Duration::from_secs(1), task).await;
match result {
Ok(Ok(())) => {
let remaining_resources = resource_count.load(Ordering::Acquire);
let cleanup_performed = cleanup_called.load(Ordering::Acquire);
if remaining_resources == 0 && cleanup_performed {
checkpoint("Resource cleanup test passed", serde_json::json!({}));
TestResult::passed()
} else {
TestResult::failed(format!(
"Resource cleanup failed: {} remaining, cleanup called: {}",
remaining_resources, cleanup_performed
))
}
}
_ => TestResult::failed("Resource cleanup test failed"),
}
})
},
)
}