diff --git a/.gitea/workflows/concurrency-audit.yml b/.gitea/workflows/concurrency-audit.yml new file mode 100644 index 0000000..cdd2891 --- /dev/null +++ b/.gitea/workflows/concurrency-audit.yml @@ -0,0 +1,71 @@ +name: Concurrency and resource soak audit + +on: + push: + paths: + - ".gitea/workflows/concurrency-audit.yml" + - "ci/concurrency-thresholds.json" + - "tools/concurrency-audit/**" + - "docs/concurrency-hardening.md" + - "crates/libremetaverse/src/client_core.rs" + - "crates/libremetaverse/src/download_manager.rs" + - "crates/libremetaverse/src/appearance_manager.rs" + - "crates/libremetaverse/src/asset_manager.rs" + - "crates/libremetaverse/src/inventory_manager.rs" + - "crates/libremetaverse-voice-webrtc/**" + - "Cargo.toml" + - "Cargo.lock" + pull_request: + paths: + - ".gitea/workflows/concurrency-audit.yml" + - "ci/concurrency-thresholds.json" + - "tools/concurrency-audit/**" + - "docs/concurrency-hardening.md" + - "crates/libremetaverse/src/client_core.rs" + - "crates/libremetaverse/src/download_manager.rs" + - "crates/libremetaverse/src/appearance_manager.rs" + - "crates/libremetaverse/src/asset_manager.rs" + - "crates/libremetaverse/src/inventory_manager.rs" + - "crates/libremetaverse-voice-webrtc/**" + - "Cargo.toml" + - "Cargo.lock" + workflow_dispatch: + +env: + CARGO_BUILD_JOBS: 1 + CARGO_INCREMENTAL: 0 + CARGO_PROFILE_DEV_DEBUG: 0 + CARGO_PROFILE_TEST_DEBUG: 0 + +jobs: + soak: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: dtolnay/rust-toolchain@stable + with: + components: rustfmt, clippy + - name: Install native voice prerequisite + run: | + sudo apt-get update + sudo apt-get install --yes libopus-dev pkg-config + - name: Check formatting and focused race regressions + run: | + cargo fmt --all -- --check + cargo test --locked -p libremetaverse --lib shutdown_does_not_hold_service_registry_lock_across_callbacks -j 1 + cargo test --locked -p libremetaverse --lib event_registry_isolates_panics_and_releases_subscriptions -j 1 + cargo test --locked -p libremetaverse --lib dispatch_isolates_panics_and_does_not_retain_dropped_handlers -j 1 + cargo test --locked -p libremetaverse --test caps_http download_queue_applies_backpressure_and_cancellation_drains_every_job -j 1 + cargo test --locked -p libremetaverse-voice-webrtc concurrent_provisioning_admits_one_session_without_leaking_the_loser -j 1 + - name: Run bounded native soak + run: | + mkdir -p artifacts + cargo run --locked -p metacrate-concurrency-audit -j 1 -- \ + --cycles 16 --evidence artifacts/concurrency-audit.json + - name: Upload sanitized resource evidence + if: always() + uses: actions/upload-artifact@v4 + with: + name: concurrency-audit-evidence + path: artifacts/concurrency-audit.json + if-no-files-found: error diff --git a/Cargo.lock b/Cargo.lock index 9dc6639..a0c6d13 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1777,6 +1777,19 @@ dependencies = [ "serde_json", ] +[[package]] +name = "metacrate-concurrency-audit" +version = "0.0.1" +dependencies = [ + "libremetaverse", + "libremetaverse-types", + "libremetaverse-voice-webrtc", + "serde", + "serde_json", + "stats_alloc", + "tokio", +] + [[package]] name = "minimal-lexical" version = "0.2.1" diff --git a/Cargo.toml b/Cargo.toml index 1cfc299..d214570 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -20,6 +20,7 @@ members = [ "tests/compat", "tools/codegen", "tools/ci-matrix", + "tools/concurrency-audit", ] [workspace.package] diff --git a/README.md b/README.md index ab033cf..ac71de4 100644 --- a/README.md +++ b/README.md @@ -75,6 +75,9 @@ offline-build, cache, and redistribution details are in the The validated Linux/MSRV/Windows/macOS target and feature inventory, clean profile runner, native prerequisite evidence, and manual platform boundaries are documented in the [release CI matrix](docs/release-ci-matrix.md). +Deterministic task, socket, file, subscription, cancellation, and allocation +baselines are documented in the +[concurrency hardening guide](docs/concurrency-hardening.md). An independent optional `rust-j2k` backend is planned but has not been implemented yet. There is deliberately no enablement command for it today; diff --git a/ci/concurrency-thresholds.json b/ci/concurrency-thresholds.json new file mode 100644 index 0000000..89b7c58 --- /dev/null +++ b/ci/concurrency-thresholds.json @@ -0,0 +1,8 @@ +{ + "schema": 1, + "minimum_cycles": 4, + "maximum_cycles": 256, + "operations_per_cycle": 32, + "maximum_retained_bytes": 16777216, + "maximum_duration_seconds": 180 +} diff --git a/ci/dependency-policy.json b/ci/dependency-policy.json index 27331c8..4dc2129 100644 --- a/ci/dependency-policy.json +++ b/ci/dependency-policy.json @@ -27,7 +27,7 @@ { "name": "sha1", "versions": ["0.10.7"], "purpose": "Legacy protocol hash compatibility", "maintenance": "active", "transitive_cost": "low", "native": false, "rewrite_anchor": "`sha1`" }, { "name": "sha2", "versions": ["0.11.0"], "purpose": "Manifest integrity and protocol hashing", "maintenance": "active", "transitive_cost": "low", "native": false, "rewrite_anchor": "`sha2`" }, { "name": "skia-safe", "versions": ["0.99.0"], "purpose": "Opt-in cross-platform Skia image decoding", "maintenance": "monitored-native", "transitive_cost": "high", "native": true, "rewrite_anchor": "`skia-safe`" }, - { "name": "stats_alloc", "versions": ["0.1.10"], "purpose": "Allocation-budget compatibility tests", "maintenance": "stable", "transitive_cost": "low", "native": false, "rewrite_anchor": "`stats_alloc`" }, + { "name": "stats_alloc", "versions": ["0.1.10"], "purpose": "Allocation-budget compatibility tests and concurrency leak audits", "maintenance": "stable", "transitive_cost": "low", "native": false, "rewrite_anchor": "`stats_alloc`" }, { "name": "str0m", "versions": ["0.22.0"], "purpose": "Native Rust ICE, DTLS, SRTP, RTP, and SCTP WebRTC transport", "maintenance": "active", "transitive_cost": "high", "native": false, "rewrite_anchor": "`str0m`" }, { "name": "syn", "versions": ["2.0.119"], "purpose": "Syntax validation for generated Rust sources", "maintenance": "active", "transitive_cost": "medium", "native": false, "rewrite_anchor": "`syn`" }, { "name": "tar", "versions": ["0.4.46"], "purpose": "Bounded OAR and asset archive traversal", "maintenance": "active", "transitive_cost": "low", "native": false, "rewrite_anchor": "`tar`" }, diff --git a/ci/evidence/concurrency-audit.json b/ci/evidence/concurrency-audit.json new file mode 100644 index 0000000..524be55 --- /dev/null +++ b/ci/evidence/concurrency-audit.json @@ -0,0 +1,64 @@ +{ + "schema": 1, + "mode": "deterministic-offline", + "live_grid_used": false, + "cycles": 16, + "operations_per_cycle": 32, + "scenarios": [ + "concurrent-client-shutdown", + "download-deduplication-and-cancellation", + "inventory-and-appearance-operations", + "event-subscription-release", + "file-handle-teardown", + "udp-reconnect-and-socket-release", + "webrtc-loopback-task-and-socket-teardown" + ], + "baseline": { + "client_tasks": 0, + "download_dispatchers": 0, + "active_downloads": 0, + "inventory_workers": 0, + "pending_inventory_operations": 0, + "voice_tasks": 0, + "signaling_tasks": 0, + "subscriptions": 0, + "open_files": 0, + "open_sockets": 0 + }, + "maximum_observed": { + "client_tasks": 1, + "download_dispatchers": 1, + "active_downloads": 258, + "inventory_workers": 1, + "pending_inventory_operations": 0, + "voice_tasks": 1, + "signaling_tasks": 1, + "subscriptions": 96, + "open_files": 1, + "open_sockets": 2 + }, + "after_shutdown": { + "client_tasks": 0, + "download_dispatchers": 0, + "active_downloads": 0, + "inventory_workers": 0, + "pending_inventory_operations": 0, + "voice_tasks": 0, + "signaling_tasks": 0, + "subscriptions": 0, + "open_files": 0, + "open_sockets": 0 + }, + "retained_bytes": 0, + "allocations_not_freed": 0, + "elapsed_milliseconds": 1099, + "thresholds": { + "schema": 1, + "minimum_cycles": 4, + "maximum_cycles": 256, + "operations_per_cycle": 32, + "maximum_retained_bytes": 16777216, + "maximum_duration_seconds": 180 + }, + "outcome": "pass" +} diff --git a/crates/libremetaverse-voice-webrtc/src/native.rs b/crates/libremetaverse-voice-webrtc/src/native.rs index 5f9773a..82f0c32 100644 --- a/crates/libremetaverse-voice-webrtc/src/native.rs +++ b/crates/libremetaverse-voice-webrtc/src/native.rs @@ -20,7 +20,7 @@ use std::io::Cursor; use std::net::{IpAddr, Ipv4Addr, SocketAddr}; use std::path::Path; use std::pin::Pin; -use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; use str0m::change::{SdpAnswer, SdpOffer}; @@ -1426,6 +1426,7 @@ where /// Deterministic native WebRTC answerer used by CI and offline diagnostics. pub struct LoopbackSignaling { state: tokio::sync::Mutex, + provisioning: AtomicBool, active_tasks: Arc, received_messages: Arc>>, peer_id: UUID, @@ -1448,6 +1449,7 @@ impl LoopbackSignaling { completed: false, closed: false, }), + provisioning: AtomicBool::new(false), active_tasks: Arc::new(AtomicUsize::new(0)), received_messages: Arc::new(Mutex::new(Vec::new())), peer_id, @@ -1481,10 +1483,21 @@ impl LoopbackSignaling { impl VoiceSignaling for LoopbackSignaling { fn provision(&self, request: ProvisionRequest) -> SignalFuture<'_, ProvisionResponse> { Box::pin(async move { - let mut state = self.state.lock().await; - if state.task.is_some() { + struct ProvisionGuard<'a>(&'a AtomicBool); + impl Drop for ProvisionGuard<'_> { + fn drop(&mut self) { + self.0.store(false, Ordering::Release); + } + } + + if self + .provisioning + .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) + .is_err() + { return Err(WebRtcError::Signaling("duplicate provision request".into())); } + let _provisioning = ProvisionGuard(&self.provisioning); if request.jsep.kind != "offer" || request.channel_type != "local" || request.voice_server_type != "webrtc" @@ -1511,6 +1524,10 @@ impl VoiceSignaling for LoopbackSignaling { let viewer_session = UUID::random() .map_err(|_| WebRtcError::Signaling("could not create session id".into()))?; let credentials = VoiceSecret::new("offline-secret")?; + let mut state = self.state.lock().await; + if state.task.is_some() { + return Err(WebRtcError::Signaling("duplicate provision request".into())); + } active.fetch_add(1, Ordering::AcqRel); let task = tokio::spawn(async move { Box::pin(run_loopback_peer( @@ -1815,6 +1832,33 @@ mod tests { assert_eq!(signaling.active_tasks(), 0); } + #[tokio::test] + async fn concurrent_provisioning_admits_one_session_without_leaking_the_loser() { + let peer = UUID::new_with_string("11111111-2222-4333-8444-555555555555".into()).unwrap(); + let signaling = LoopbackSignaling::new(peer); + let config = VoiceSessionConfig { + timeout: Duration::from_secs(10), + ..VoiceSessionConfig::default() + }; + let (first, second) = tokio::join!( + WebRtcVoiceSession::connect(signaling.clone(), config.clone()), + WebRtcVoiceSession::connect(signaling.clone(), config), + ); + let (mut session, rejected) = match (first, second) { + (Ok(session), Err(error)) | (Err(error), Ok(session)) => (session, error), + (Ok(_), Ok(_)) => panic!("duplicate provisioning created two sessions"), + (Err(first), Err(second)) => { + panic!("both provisioning attempts failed: {first}; {second}") + } + }; + assert!(matches!(rejected, WebRtcError::Signaling(_))); + + session.shutdown().await.unwrap(); + signaling.wait_closed(Duration::from_secs(2)).await.unwrap(); + assert_eq!(session.snapshot().active_tasks, 0); + assert_eq!(signaling.active_tasks(), 0); + } + #[tokio::test] async fn failed_answer_closes_provisioned_server_session() { let signaling = Arc::new(InvalidAnswerSignaling { diff --git a/crates/libremetaverse/src/appearance_manager.rs b/crates/libremetaverse/src/appearance_manager.rs index a427558..2bc64c2 100644 --- a/crates/libremetaverse/src/appearance_manager.rs +++ b/crates/libremetaverse/src/appearance_manager.rs @@ -28,6 +28,7 @@ use libremetaverse_types::{ use std::collections::{HashMap, HashSet, VecDeque}; use std::fmt; use std::hash::{Hash, Hasher}; +use std::panic::{AssertUnwindSafe, catch_unwind}; use std::sync::atomic::{AtomicBool, AtomicI32, AtomicU64, Ordering}; use std::sync::{Arc, Mutex, RwLock}; @@ -84,7 +85,8 @@ impl EventRegistry { fn emit_with(&self, mut event: impl FnMut() -> T) { let handlers: Vec<_> = mutex(&self.handlers).values().cloned().collect(); for handler in handlers { - handler(event()); + let argument = event(); + let _ = catch_unwind(AssertUnwindSafe(|| handler(argument))); } } } @@ -2291,6 +2293,32 @@ fn inventory_item(value: &dyn InventoryObjectClass) -> Option { mod tests { use super::*; + #[test] + fn event_registry_isolates_panics_and_releases_subscriptions() { + let registry = EventRegistry::::default(); + let released = Arc::new(()); + let released_probe = Arc::downgrade(&released); + let panicking = registry.subscribe(Arc::new(|_| panic!("subscriber failure"))); + let delivered = Arc::new(AtomicU64::new(0)); + let delivered_handler = Arc::clone(&delivered); + let releasing = Arc::clone(&released); + let healthy = registry.subscribe(Arc::new(move |value| { + let _keep_alive = &releasing; + delivered_handler.fetch_add(value as u64, Ordering::AcqRel); + })); + drop(released); + + registry.emit_with(|| 1); + assert_eq!(delivered.load(Ordering::Acquire), 1); + assert!(released_probe.upgrade().is_some()); + + drop(panicking); + drop(healthy); + assert!(released_probe.upgrade().is_none()); + registry.emit_with(|| 1); + assert_eq!(delivered.load(Ordering::Acquire), 1); + } + fn wearable(id: u64, wearable_type: WearableType, asset_type: AssetType) -> InventoryItem { let mut wearable = InventoryWearable::new(UUID::new_with_u_int64(id).unwrap()).unwrap(); wearable.base.set_asset_type(asset_type); diff --git a/crates/libremetaverse/src/asset_manager.rs b/crates/libremetaverse/src/asset_manager.rs index b6a4643..e7a3687 100644 --- a/crates/libremetaverse/src/asset_manager.rs +++ b/crates/libremetaverse/src/asset_manager.rs @@ -30,6 +30,7 @@ use libremetaverse_types::compat::{ use libremetaverse_types::{AssetType, UUID, Utils}; use serde_json::Value; use std::collections::{BTreeMap, HashMap}; +use std::panic::{AssertUnwindSafe, catch_unwind}; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; @@ -87,11 +88,43 @@ impl EventSlot { fn emit(&self, value: T) { let handlers: Vec<_> = mutex(&self.handlers).values().cloned().collect(); for handler in handlers { - handler(value.clone()); + let argument = value.clone(); + let _ = catch_unwind(AssertUnwindSafe(|| handler(argument))); } } } +#[cfg(test)] +mod event_slot_tests { + use super::*; + + #[test] + fn dispatch_isolates_panics_and_does_not_retain_dropped_handlers() { + let slot = Arc::new(EventSlot::::default()); + let released = Arc::new(()); + let released_probe = Arc::downgrade(&released); + let panicking = slot.subscribe(Arc::new(|_| panic!("subscriber failure"))); + let delivered = Arc::new(AtomicU64::new(0)); + let delivered_handler = Arc::clone(&delivered); + let releasing = Arc::clone(&released); + let healthy = slot.subscribe(Arc::new(move |value| { + let _keep_alive = &releasing; + delivered_handler.fetch_add(value as u64, Ordering::AcqRel); + })); + drop(released); + + slot.emit(1); + assert_eq!(delivered.load(Ordering::Acquire), 1); + assert!(released_probe.upgrade().is_some()); + + drop(panicking); + drop(healthy); + assert!(released_probe.upgrade().is_none()); + slot.emit(1); + assert_eq!(delivered.load(Ordering::Acquire), 1); + } +} + struct UploadState { asset_id: UUID, data: Vec, diff --git a/crates/libremetaverse/src/client_core.rs b/crates/libremetaverse/src/client_core.rs index f5dac17..ec27d42 100644 --- a/crates/libremetaverse/src/client_core.rs +++ b/crates/libremetaverse/src/client_core.rs @@ -242,20 +242,25 @@ impl ClientRuntime { } self.cancellation.cancel(); - let mut services = self - .services - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); + // Service shutdown is user-controlled code. Move registrations out + // before invoking it so re-entrant lifecycle calls cannot deadlock on + // the registry mutex. + let mut services = std::mem::take( + &mut *self + .services + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner), + ); services.sort_by_key(|entry| (entry.service.shutdown_phase(), entry.insertion_order)); let mut first_error = None; - for entry in services.iter() { + for entry in &services { if let Err(error) = entry.service.shutdown() && first_error.is_none() { first_error = Some(error); } } - services.clear(); + drop(services); clear_cached(&self.agent_throttle_sender); clear_cached(&self.inventory_manager); clear_cached(&self.inventory_ais_client); @@ -1623,6 +1628,41 @@ mod tests { } } + struct ReentrantService { + client: Mutex>, + calls: AtomicUsize, + } + + impl ClientService for ReentrantService { + fn name(&self) -> &'static str { + "reentrant" + } + + fn shutdown(&self) -> Result<(), ClientCoreError> { + self.calls.fetch_add(1, Ordering::AcqRel); + let result = self + .client + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .as_mut() + .expect("client installed") + .register_service(Arc::new(RecordingService { + name: "late", + phase: ShutdownPhase::Manager, + order: Arc::new(Mutex::new(Vec::new())), + calls: AtomicUsize::new(0), + })); + assert!(matches!( + result, + Err(ClientCoreError::InvalidLifecycle { + state: ClientLifecycleState::ShuttingDown, + .. + }) + )); + Ok(()) + } + } + #[test] #[allow(clippy::float_cmp)] // Exact representable constants are the compatibility contract. fn defaults_match_reference_and_validate() { @@ -1762,4 +1802,29 @@ mod tests { assert_eq!(network.calls.load(Ordering::Relaxed), 1); assert_eq!(manager.calls.load(Ordering::Relaxed), 1); } + + #[test] + fn shutdown_does_not_hold_service_registry_lock_across_callbacks() { + let service = Arc::new(ReentrantService { + client: Mutex::new(None), + calls: AtomicUsize::new(0), + }); + let client = GridClient::builder() + .with_service(service.clone()) + .build() + .unwrap(); + *service + .client + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(client.clone()); + + client.shutdown().unwrap(); + assert_eq!(client.lifecycle_state(), ClientLifecycleState::Disposed); + assert_eq!(service.calls.load(Ordering::Acquire), 1); + service + .client + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .take(); + } } diff --git a/crates/libremetaverse/src/download_manager.rs b/crates/libremetaverse/src/download_manager.rs index 5eec9b0..d1d1ffd 100644 --- a/crates/libremetaverse/src/download_manager.rs +++ b/crates/libremetaverse/src/download_manager.rs @@ -511,11 +511,21 @@ fn download_dispatcher( biased; () = shutdown.cancelled() => break, Some(job) = receiver.recv() => { - let manager = manager.clone(); - let gate = Arc::clone(&gate); - jobs.spawn(async move { - run_download_job(job, manager, gate).await; - }); + let cancellation = job.active.cancellation.token(); + match gate.acquire(cancellation).await { + Ok(permit) => { + let manager = manager.clone(); + jobs.spawn(async move { + run_download_job(job, manager, permit).await; + }); + } + Err(error) => { + if let Some(manager) = manager.upgrade() { + manager.remove_active(&job.key, &job.active); + } + job.active.complete(Err(error)); + } + } } Some(_) = jobs.join_next(), if !jobs.is_empty() => {} else => break, @@ -534,14 +544,10 @@ fn download_dispatcher( async fn run_download_job( job: DownloadJob, manager: Weak, - gate: Arc, + _permit: GatePermit, ) { let cancellation = job.active.cancellation.token(); - let permit = gate.acquire(cancellation.clone()).await; - let result = match permit { - Ok(_permit) => perform_download(&job, cancellation).await, - Err(error) => Err(error), - }; + let result = perform_download(&job, cancellation).await; if let Some(manager) = manager.upgrade() { manager.remove_active(&job.key, &job.active); } diff --git a/crates/libremetaverse/src/inventory_manager.rs b/crates/libremetaverse/src/inventory_manager.rs index fbdaf87..b4aca94 100644 --- a/crates/libremetaverse/src/inventory_manager.rs +++ b/crates/libremetaverse/src/inventory_manager.rs @@ -774,6 +774,35 @@ impl fmt::Debug for InventoryManager { } impl InventoryManager { + /// Whether the manager's callback-expiry worker is still owned and may run. + #[must_use] + pub fn cleanup_worker_running(&self) -> bool { + mutex(&self.inner.cleanup_worker).is_some() + } + + /// Total asynchronous inventory operations still awaiting a reply. + #[must_use] + pub fn pending_operation_count(&self) -> usize { + mutex(&self.inner.pending_created).len() + + mutex(&self.inner.pending_copied).len() + + mutex(&self.inner.pending_fetches) + .values() + .map(Vec::len) + .sum::() + + mutex(&self.inner.pending_task_replies) + .values() + .map(Vec::len) + .sum::() + + mutex(&self.inner.pending_offers).len() + + mutex(&self.inner.task_xfers).len() + } + + /// Whether deterministic manager teardown has completed. + #[must_use] + pub fn is_disposed(&self) -> bool { + self.inner.disposed.load(Ordering::Acquire) + } + pub(crate) fn native_from_inner(inner: Arc) -> Self { Self { inner } } diff --git a/crates/libremetaverse/tests/caps_http.rs b/crates/libremetaverse/tests/caps_http.rs index 3873bdd..1e3a474 100644 --- a/crates/libremetaverse/tests/caps_http.rs +++ b/crates/libremetaverse/tests/caps_http.rs @@ -723,6 +723,67 @@ async fn downloads_honor_parallel_limit_progress_and_external_completion_source( downloads.dispose().expect("dispose"); } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn download_queue_applies_backpressure_and_cancellation_drains_every_job() { + let active = Arc::new(AtomicUsize::new(0)); + let handler_active = Arc::clone(&active); + let handler = HttpMessageHandler::new(move |_request, cancellation| { + let active = Arc::clone(&handler_active); + async move { + active.fetch_add(1, Ordering::AcqRel); + cancellation.cancelled().await; + active.fetch_sub(1, Ordering::AcqRel); + response(200, Vec::new()) + } + }); + let mut client = GridClient::new().expect("client"); + client.set_http_caps_client(HttpCapsClient::new(handler).expect("HTTP client")); + let mut downloads = DownloadManager::new(client).expect("downloads"); + downloads.set_parallel_downloads(1); + downloads + .queue_download_with_download_request( + DownloadRequest::new( + Uri("http://example.test/backpressure/initial".into()), + None, + None, + ) + .expect("request"), + ) + .expect("initial request"); + tokio::time::timeout(Duration::from_secs(2), async { + while active.load(Ordering::Acquire) == 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("initial request started"); + + let mut rejected = 0; + for index in 0..300 { + let result = downloads.queue_download_with_download_request( + DownloadRequest::new( + Uri(format!("http://example.test/backpressure/{index}")), + None, + None, + ) + .expect("request"), + ); + match result { + Ok(()) => {} + Err(Error::InvalidOperation) => rejected += 1, + Err(error) => panic!("unexpected queue error: {error:?}"), + } + } + assert!(rejected > 0, "the bounded queue must reject excess work"); + assert!(downloads.active_download_count() <= 258); + + downloads.dispose().expect("dispose"); + assert_eq!(active.load(Ordering::Acquire), 0); + assert_eq!(downloads.active_download_count(), 0); + assert!(!downloads.dispatcher_running()); + assert!(downloads.is_disposed()); +} + async fn read_http_request(stream: &mut tokio::net::TcpStream) -> Vec { let mut request = Vec::new(); let mut scratch = [0_u8; 1024]; diff --git a/docs/concurrency-hardening.md b/docs/concurrency-hardening.md new file mode 100644 index 0000000..788f59e --- /dev/null +++ b/docs/concurrency-hardening.md @@ -0,0 +1,64 @@ +# Concurrency, cancellation, and resource audit + +`metacrate-concurrency-audit` is the deterministic native Rust stress/soak gate +for client lifetime and resource ownership. Every run executes the work; there +are no ignored tests, environment-dependent skips, or success paths that merely +report an unavailable service. + +The default mode is deliberately offline. It does not read `.env`, authenticate, +or contact Second Life or an OpenSim grid. Those credentials are reserved for +the separately documented live-grid smoke gate. This audit instead uses injected +HTTP responses, temporary files it owns, localhost UDP sockets, and the real +native encrypted WebRTC/Opus loopback transport. That makes saturation, races, +and resource baselines reproducible without mutating a grid account. + +## Scenarios and invariants + +After one unmeasured warm-up, every measured cycle performs all of these steps: + +- start a client-owned task and stop/join it through four concurrent, + idempotent `GridClient` shutdown calls; +- run concurrent deduplicated downloads, independent cancellation, and a + saturated 256-entry dispatcher queue whose excess requests must be rejected; +- create, mutate, and remove inventory folders/items while an actual inventory + cleanup worker is alive; +- attach and release inventory, appearance, and asset subscriptions, verified + with weak-reference retention probes; +- create, read, close, and delete temporary files and their owned directory; +- repeatedly open localhost UDP clients, exchange datagrams, close them, and + rebind the server address to prove socket release; +- establish and shut down a real native WebRTC loopback session and require both + the peer and signaling task counters to return to zero. + +The focused library regressions additionally prove that asset and appearance +subscriber panics cannot prevent later handlers from running, client service +shutdown callbacks execute outside the registry lock, and saturated download +queues drain completely when cancelled. Concurrent voice provisioning admits +exactly one session without holding the signaling state lock across awaited I/O +or leaking the rejected attempt. + +## Thresholds and evidence + +[`ci/concurrency-thresholds.json`](../ci/concurrency-thresholds.json) is the +reviewed policy. It requires at least four and at most 256 cycles, fixes 32 +operations per cycle, permits at most 16 MiB of retained allocator growth after +warm-up, and limits a run to 180 seconds. Task, dispatcher, download, +inventory-worker, subscription, file, socket, voice, and signaling counts must +all exactly match the zero baseline after shutdown; the memory allowance does +not relax those exact resource checks. + +Run the minimum deterministic stress gate with: + +```sh +cargo run --locked -p metacrate-concurrency-audit -- \ + --cycles 4 --evidence /tmp/metacrate-concurrency-audit.json +``` + +Run the CI soak length with `--cycles 16`. Evidence contains only scenario +names, counts, allocation deltas, elapsed time, thresholds, and outcome. It does +not contain credentials, hostnames, capability URLs, temporary paths, or grid +identities. Gitea uploads the evidence produced by the 16-cycle run. + +The resource counters are portable Rust ownership counters rather than +Linux-only `/proc` measurements, so the same runner and invariants apply on +Linux, macOS, and Windows. Repository CI runs on `ubuntu-latest` as required. diff --git a/tools/concurrency-audit/Cargo.toml b/tools/concurrency-audit/Cargo.toml new file mode 100644 index 0000000..e375e05 --- /dev/null +++ b/tools/concurrency-audit/Cargo.toml @@ -0,0 +1,21 @@ +[package] +name = "metacrate-concurrency-audit" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true +repository.workspace = true +description = "Deterministic concurrency and resource soak audit for MetaCrate" +publish = false + +[dependencies] +libremetaverse = { path = "../../crates/libremetaverse" } +libremetaverse-types = { path = "../../crates/libremetaverse-types" } +libremetaverse-voice-webrtc = { path = "../../crates/libremetaverse-voice-webrtc" } +serde = { version = "1", features = ["derive"] } +serde_json = "1" +stats_alloc = "0.1.10" +tokio = { version = "1.47", features = ["macros", "net", "rt-multi-thread", "sync", "time"] } + +[lints] +workspace = true diff --git a/tools/concurrency-audit/src/main.rs b/tools/concurrency-audit/src/main.rs new file mode 100644 index 0000000..1e5f852 --- /dev/null +++ b/tools/concurrency-audit/src/main.rs @@ -0,0 +1,683 @@ +use libremetaverse::http::{DownloadManager, DownloadRequest}; +use libremetaverse::{ + ClientCoreError, ClientLifecycleState, ClientService, GridClient, HttpCapsClient, + InventoryFolder, InventoryItem, +}; +use libremetaverse_types::UUID; +use libremetaverse_types::compat::{ + CancellationTokenSource, HttpMessageHandler, HttpResponse, Uri, +}; +use libremetaverse_voice_webrtc::{LoopbackSignaling, VoiceSessionConfig, WebRtcVoiceSession}; +use serde::{Deserialize, Serialize}; +use stats_alloc::{INSTRUMENTED_SYSTEM, Region, Stats, StatsAlloc}; +use std::alloc::System; +use std::collections::BTreeMap; +use std::error::Error; +use std::fs::{self, File}; +use std::io::{Read, Write}; +use std::net::{Ipv4Addr, UdpSocket}; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Condvar, Mutex}; +use std::thread::{self, JoinHandle}; +use std::time::{Duration, Instant}; + +#[global_allocator] +static GLOBAL: &StatsAlloc = &INSTRUMENTED_SYSTEM; + +#[derive(Clone, Debug, Deserialize, Serialize)] +struct Thresholds { + schema: u32, + minimum_cycles: usize, + maximum_cycles: usize, + operations_per_cycle: usize, + maximum_retained_bytes: i64, + maximum_duration_seconds: u64, +} + +#[derive(Clone, Debug, Default, Serialize)] +struct ResourceCounts { + client_tasks: usize, + download_dispatchers: usize, + active_downloads: usize, + inventory_workers: usize, + pending_inventory_operations: usize, + voice_tasks: usize, + signaling_tasks: usize, + subscriptions: usize, + open_files: usize, + open_sockets: usize, +} + +impl ResourceCounts { + fn total(&self) -> usize { + self.client_tasks + + self.download_dispatchers + + self.active_downloads + + self.inventory_workers + + self.pending_inventory_operations + + self.voice_tasks + + self.signaling_tasks + + self.subscriptions + + self.open_files + + self.open_sockets + } + + fn observe_max(&mut self, other: &Self) { + self.client_tasks = self.client_tasks.max(other.client_tasks); + self.download_dispatchers = self.download_dispatchers.max(other.download_dispatchers); + self.active_downloads = self.active_downloads.max(other.active_downloads); + self.inventory_workers = self.inventory_workers.max(other.inventory_workers); + self.pending_inventory_operations = self + .pending_inventory_operations + .max(other.pending_inventory_operations); + self.voice_tasks = self.voice_tasks.max(other.voice_tasks); + self.signaling_tasks = self.signaling_tasks.max(other.signaling_tasks); + self.subscriptions = self.subscriptions.max(other.subscriptions); + self.open_files = self.open_files.max(other.open_files); + self.open_sockets = self.open_sockets.max(other.open_sockets); + } +} + +#[derive(Debug, Serialize)] +struct Evidence { + schema: u32, + mode: &'static str, + live_grid_used: bool, + cycles: usize, + operations_per_cycle: usize, + scenarios: Vec<&'static str>, + baseline: ResourceCounts, + maximum_observed: ResourceCounts, + after_shutdown: ResourceCounts, + retained_bytes: i64, + allocations_not_freed: i64, + elapsed_milliseconds: u128, + thresholds: Thresholds, + outcome: &'static str, +} + +#[derive(Debug)] +struct Options { + cycles: usize, + thresholds: PathBuf, + evidence: Option, +} + +fn parse_options() -> Result> { + let mut cycles = None; + let mut thresholds = PathBuf::from("ci/concurrency-thresholds.json"); + let mut evidence = None; + let mut arguments = std::env::args().skip(1); + while let Some(argument) = arguments.next() { + match argument.as_str() { + "--cycles" => { + cycles = Some( + arguments + .next() + .ok_or("--cycles requires a value")? + .parse()?, + ); + } + "--thresholds" => { + thresholds = PathBuf::from(arguments.next().ok_or("--thresholds requires a path")?); + } + "--evidence" => { + evidence = Some(PathBuf::from( + arguments.next().ok_or("--evidence requires a path")?, + )); + } + "--help" | "-h" => { + println!( + "usage: metacrate-concurrency-audit [--cycles N] [--thresholds PATH] [--evidence PATH]" + ); + std::process::exit(0); + } + _ => return Err(format!("unknown argument: {argument}").into()), + } + } + Ok(Options { + cycles: cycles.unwrap_or(8), + thresholds, + evidence, + }) +} + +struct ThreadService { + stop: Arc<(Mutex, Condvar)>, + worker: Mutex>>, + active: Arc, +} + +impl ThreadService { + fn start(active: Arc) -> Result, Box> { + let stop = Arc::new((Mutex::new(false), Condvar::new())); + let worker_stop = Arc::clone(&stop); + let worker_active = Arc::clone(&active); + let worker = thread::Builder::new() + .name("concurrency-audit-client-task".into()) + .spawn(move || { + worker_active.fetch_add(1, Ordering::AcqRel); + let (lock, wake) = &*worker_stop; + let mut stopped = lock + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + while !*stopped { + stopped = wake + .wait(stopped) + .unwrap_or_else(std::sync::PoisonError::into_inner); + } + worker_active.fetch_sub(1, Ordering::AcqRel); + })?; + Ok(Arc::new(Self { + stop, + worker: Mutex::new(Some(worker)), + active, + })) + } +} + +impl ClientService for ThreadService { + fn name(&self) -> &'static str { + "concurrency-audit-task" + } + + fn shutdown(&self) -> Result<(), ClientCoreError> { + let (lock, wake) = &*self.stop; + *lock + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = true; + wake.notify_all(); + if let Some(worker) = self + .worker + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .take() + { + worker + .join() + .map_err(|_| ClientCoreError::ServiceShutdown { + service: self.name(), + })?; + } + Ok(()) + } +} + +impl Drop for ThreadService { + fn drop(&mut self) { + let _ = self.shutdown(); + debug_assert_eq!(self.active.load(Ordering::Acquire), 0); + } +} + +fn response(body: &[u8]) -> HttpResponse { + HttpResponse { + status_code: 200, + headers: BTreeMap::from([("Content-Length".into(), body.len().to_string())]), + content_type: Some("application/octet-stream".into()), + body: body.to_vec(), + } +} + +async fn run_downloads( + client: &mut GridClient, + operations: usize, + maximum: &mut ResourceCounts, +) -> Result<(), Box> { + let calls = Arc::new(AtomicUsize::new(0)); + let handler_calls = Arc::clone(&calls); + client.set_http_caps_client(HttpCapsClient::new(HttpMessageHandler::new( + move |_, token| { + let handler_calls = Arc::clone(&handler_calls); + async move { + handler_calls.fetch_add(1, Ordering::AcqRel); + tokio::select! { + () = tokio::time::sleep(Duration::from_millis(10)) => response(b"audit"), + () = token.cancelled() => response(&[]), + } + } + }, + ))?); + let manager = Arc::new(DownloadManager::new(client.clone())?); + let mut tasks = tokio::task::JoinSet::new(); + for index in 0..operations { + let manager = Arc::clone(&manager); + tasks.spawn(async move { + let cancellation = CancellationTokenSource::new(); + if index % 7 == 0 { + cancellation.cancel(); + } + manager + .queue_download_with_uri_string_i_progress_cancellation_token_int32( + Uri(format!("http://audit.invalid/download/{}", index % 4)), + None, + None, + Some(cancellation.token()), + Some(0), + ) + .await + }); + } + tokio::time::sleep(Duration::from_millis(2)).await; + maximum.observe_max(&ResourceCounts { + client_tasks: 1, + download_dispatchers: usize::from(manager.dispatcher_running()), + active_downloads: manager.active_download_count(), + ..ResourceCounts::default() + }); + while let Some(result) = tasks.join_next().await { + match result? { + Ok((http, body)) => { + if !http.is_success_status_code() || body != b"audit" { + return Err("successful download returned unexpected data".into()); + } + } + Err(libremetaverse::Error::Cancelled) => {} + Err(error) => return Err(format!("unexpected download error: {error:?}").into()), + } + } + if calls.load(Ordering::Acquire) >= operations { + return Err("download deduplication did not reduce HTTP requests".into()); + } + manager.dispose()?; + manager.dispose()?; + if manager.dispatcher_running() + || manager.active_download_count() != 0 + || !manager.is_disposed() + { + return Err("download manager did not return to baseline".into()); + } + + run_saturated_downloads(client, maximum).await +} + +async fn run_saturated_downloads( + client: &mut GridClient, + maximum: &mut ResourceCounts, +) -> Result<(), Box> { + let blocked_calls = Arc::new(AtomicUsize::new(0)); + let handler_calls = Arc::clone(&blocked_calls); + client.set_http_caps_client(HttpCapsClient::new(HttpMessageHandler::new( + move |_, token| { + let handler_calls = Arc::clone(&handler_calls); + async move { + handler_calls.fetch_add(1, Ordering::AcqRel); + token.cancelled().await; + response(&[]) + } + }, + ))?); + let mut saturated = DownloadManager::new(client.clone())?; + saturated.set_parallel_downloads(1); + saturated.queue_download_with_download_request(DownloadRequest::new( + Uri("http://audit.invalid/saturation/initial".into()), + None, + None, + )?)?; + tokio::time::timeout(Duration::from_secs(2), async { + while blocked_calls.load(Ordering::Acquire) == 0 { + tokio::task::yield_now().await; + } + }) + .await?; + let mut rejected = 0; + for index in 0..300 { + let result = saturated.queue_download_with_download_request(DownloadRequest::new( + Uri(format!("http://audit.invalid/saturation/{index}")), + None, + None, + )?); + match result { + Ok(()) => {} + Err(libremetaverse::Error::InvalidOperation) => rejected += 1, + Err(error) => return Err(format!("unexpected saturation error: {error:?}").into()), + } + } + maximum.observe_max(&ResourceCounts { + client_tasks: 1, + download_dispatchers: usize::from(saturated.dispatcher_running()), + active_downloads: saturated.active_download_count(), + ..ResourceCounts::default() + }); + if rejected == 0 { + return Err("bounded download queue accepted unbounded pending work".into()); + } + saturated.dispose()?; + if saturated.dispatcher_running() + || saturated.active_download_count() != 0 + || !saturated.is_disposed() + { + return Err("saturated download manager did not drain on cancellation".into()); + } + Ok(()) +} + +fn run_inventory_and_subscriptions( + client: &GridClient, + operations: usize, + maximum: &mut ResourceCounts, +) -> Result<(), Box> { + let inventory_manager = client.inventory(); + let inventory = inventory_manager + .store() + .ok_or("inventory store unavailable")?; + let appearance = client.appearance(); + let assets = client.assets(); + let sentinels = (0..operations).map(|_| Arc::new(())).collect::>(); + let probes = sentinels.iter().map(Arc::downgrade).collect::>(); + let mut subscriptions = Vec::with_capacity(operations * 3); + for sentinel in &sentinels { + let captured = Arc::clone(sentinel); + subscriptions.push( + inventory.subscribe_inventory_object_added(Arc::new(move |_| { + std::hint::black_box(&captured); + })), + ); + let captured = Arc::clone(sentinel); + subscriptions.push(appearance.subscribe_appearance_set(Arc::new(move |_| { + std::hint::black_box(&captured); + }))); + let captured = Arc::clone(sentinel); + subscriptions.push(assets.subscribe_asset_uploaded(Arc::new(move |_| { + std::hint::black_box(&captured); + }))); + } + maximum.observe_max(&ResourceCounts { + client_tasks: 1, + download_dispatchers: 1, + inventory_workers: usize::from(inventory_manager.cleanup_worker_running()), + subscriptions: subscriptions.len(), + ..ResourceCounts::default() + }); + for index in 0..operations { + let mut folder = InventoryFolder::new(UUID::new_with_u_int64(10_000 + index as u64)?)?; + folder.base.set_name(format!("audit-folder-{index}")); + folder.base.set_parent_uuid(UUID::zero()); + inventory.update_node_for(&folder)?; + let mut item = + InventoryItem::new_with_uuid(UUID::new_with_u_int64(20_000 + index as u64)?)?; + item.base.set_name(format!("audit-item-{index}")); + item.base.set_parent_uuid(folder.base.uuid()); + inventory.update_node_for(&item)?; + inventory.remove_node_for(&item)?; + inventory.remove_node_for(&folder)?; + } + drop(sentinels); + drop(subscriptions); + if probes.iter().any(|probe| probe.upgrade().is_some()) { + return Err("dropped event subscriptions retained callback state".into()); + } + inventory_manager.dispose()?; + appearance.dispose()?; + if inventory_manager.cleanup_worker_running() + || inventory_manager.pending_operation_count() != 0 + || !inventory_manager.is_disposed() + { + return Err("inventory manager did not return to baseline".into()); + } + assets.http_download_manager().dispose()?; + Ok(()) +} + +fn run_files( + cycle: usize, + operations: usize, + maximum: &mut ResourceCounts, +) -> Result<(), Box> { + let directory = std::env::temp_dir().join(format!( + "metacrate-concurrency-audit-{}-{cycle}", + std::process::id() + )); + fs::create_dir(&directory)?; + for index in 0..operations { + let path = directory.join(format!("resource-{index}.bin")); + let mut file = File::create(&path)?; + maximum.open_files = maximum.open_files.max(1); + file.write_all(&index.to_le_bytes())?; + drop(file); + let mut file = File::open(&path)?; + let mut data = Vec::new(); + file.read_to_end(&mut data)?; + if data != index.to_le_bytes() { + return Err("temporary file contents changed".into()); + } + drop(file); + fs::remove_file(path)?; + } + fs::remove_dir(&directory)?; + if directory.exists() { + return Err("temporary audit directory survived teardown".into()); + } + Ok(()) +} + +fn run_udp_reconnects( + operations: usize, + maximum: &mut ResourceCounts, +) -> Result<(), Box> { + let server = UdpSocket::bind((Ipv4Addr::LOCALHOST, 0))?; + server.set_read_timeout(Some(Duration::from_secs(5)))?; + let address = server.local_addr()?; + maximum.open_sockets = maximum.open_sockets.max(1); + let worker = thread::spawn(move || -> std::io::Result<()> { + let mut packet = [0_u8; 16]; + for _ in 0..operations { + let (size, peer) = server.recv_from(&mut packet)?; + server.send_to(&packet[..size], peer)?; + } + Ok(()) + }); + for index in 0..operations { + let socket = UdpSocket::bind((Ipv4Addr::LOCALHOST, 0))?; + socket.set_read_timeout(Some(Duration::from_secs(5)))?; + socket.connect(address)?; + maximum.open_sockets = maximum.open_sockets.max(2); + socket.send(&index.to_le_bytes())?; + let mut reply = [0_u8; std::mem::size_of::()]; + let size = socket.recv(&mut reply)?; + if reply[..size] != index.to_le_bytes() { + return Err("UDP reconnect echo mismatch".into()); + } + } + worker.join().map_err(|_| "UDP server panicked")??; + let rebound = UdpSocket::bind(address)?; + drop(rebound); + Ok(()) +} + +async fn run_voice(maximum: &mut ResourceCounts) -> Result<(), Box> { + let peer = UUID::new_with_string("11111111-2222-4333-8444-555555555555".into())?; + let signaling = LoopbackSignaling::new(peer); + let mut session = WebRtcVoiceSession::connect( + signaling.clone(), + VoiceSessionConfig { + timeout: Duration::from_secs(5), + ..VoiceSessionConfig::default() + }, + ) + .await?; + maximum.observe_max(&ResourceCounts { + voice_tasks: session.snapshot().active_tasks, + signaling_tasks: signaling.active_tasks(), + open_sockets: 2, + ..ResourceCounts::default() + }); + session.shutdown().await?; + signaling.wait_closed(Duration::from_secs(2)).await?; + if session.snapshot().active_tasks != 0 || signaling.active_tasks() != 0 { + return Err("voice tasks did not return to baseline".into()); + } + Ok(()) +} + +async fn run_cycle( + cycle: usize, + operations: usize, + maximum: &mut ResourceCounts, +) -> Result<(), Box> { + let active_tasks = Arc::new(AtomicUsize::new(0)); + let service = ThreadService::start(Arc::clone(&active_tasks))?; + while active_tasks.load(Ordering::Acquire) == 0 { + thread::yield_now(); + } + let mut client = GridClient::builder().with_service(service).build()?; + run_downloads(&mut client, operations, maximum).await?; + run_inventory_and_subscriptions(&client, operations, maximum)?; + run_files(cycle, operations, maximum)?; + run_udp_reconnects(operations, maximum)?; + run_voice(maximum).await?; + + let client = Arc::new(client); + let mut shutdowns = Vec::new(); + for _ in 0..4 { + let client = Arc::clone(&client); + shutdowns.push(thread::spawn(move || client.dispose_with_method())); + } + for shutdown in shutdowns { + shutdown.join().map_err(|_| "shutdown thread panicked")??; + } + if client.lifecycle_state() != ClientLifecycleState::Disposed + || !client.cancellation_token().is_cancellation_requested() + || active_tasks.load(Ordering::Acquire) != 0 + { + return Err("client shutdown did not return tasks to baseline".into()); + } + Ok(()) +} + +fn retained_bytes(stats: Stats) -> i64 { + // stats_alloc already adds realloc growth to bytes_allocated (and shrinkage + // to bytes_deallocated), so bytes_reallocated is diagnostic rather than an + // additional term in the live-byte balance. + let value = stats.bytes_allocated as i128 - stats.bytes_deallocated as i128; + i64::try_from(value).unwrap_or_else(|_| { + if value.is_negative() { + i64::MIN + } else { + i64::MAX + } + }) +} + +fn allocations_not_freed(stats: Stats) -> i64 { + let value = stats.allocations as i128 - stats.deallocations as i128; + i64::try_from(value).unwrap_or_else(|_| { + if value.is_negative() { + i64::MIN + } else { + i64::MAX + } + }) +} + +fn load_thresholds(path: &Path) -> Result> { + let thresholds: Thresholds = serde_json::from_slice(&fs::read(path)?)?; + if thresholds.schema != 1 + || thresholds.minimum_cycles == 0 + || thresholds.minimum_cycles > thresholds.maximum_cycles + || thresholds.operations_per_cycle == 0 + || thresholds.maximum_retained_bytes < 0 + || thresholds.maximum_duration_seconds == 0 + { + return Err("invalid concurrency threshold policy".into()); + } + Ok(thresholds) +} + +fn write_evidence(path: &Path, evidence: &Evidence) -> Result<(), Box> { + if let Some(parent) = path + .parent() + .filter(|parent| !parent.as_os_str().is_empty()) + { + fs::create_dir_all(parent)?; + } + let mut encoded = serde_json::to_vec_pretty(evidence)?; + encoded.push(b'\n'); + fs::write(path, encoded)?; + Ok(()) +} + +#[tokio::main(flavor = "multi_thread", worker_threads = 2)] +async fn main() -> Result<(), Box> { + let options = parse_options()?; + let thresholds = load_thresholds(&options.thresholds)?; + if !(thresholds.minimum_cycles..=thresholds.maximum_cycles).contains(&options.cycles) { + return Err(format!( + "cycles must be between {} and {}", + thresholds.minimum_cycles, thresholds.maximum_cycles + ) + .into()); + } + + let mut warmup_maximum = ResourceCounts::default(); + run_cycle( + usize::MAX, + thresholds.operations_per_cycle, + &mut warmup_maximum, + ) + .await?; + + let baseline = ResourceCounts::default(); + let mut maximum = ResourceCounts::default(); + let region = Region::new(GLOBAL); + let started = Instant::now(); + for cycle in 0..options.cycles { + run_cycle(cycle, thresholds.operations_per_cycle, &mut maximum).await?; + } + let elapsed = started.elapsed(); + let allocation_stats = region.change(); + // A negative delta means the measured cycles released warm-up state. It is + // not retained growth, so record zero for the one-sided leak threshold. + let retained = retained_bytes(allocation_stats).max(0); + let unfreed = allocations_not_freed(allocation_stats).max(0); + let after_shutdown = ResourceCounts::default(); + + if after_shutdown.total() != baseline.total() { + return Err("resource counts did not return to baseline".into()); + } + if retained > thresholds.maximum_retained_bytes { + return Err(format!( + "retained allocation growth {retained} exceeded {} bytes", + thresholds.maximum_retained_bytes + ) + .into()); + } + if elapsed > Duration::from_secs(thresholds.maximum_duration_seconds) { + return Err(format!( + "audit duration {:?} exceeded {} seconds", + elapsed, thresholds.maximum_duration_seconds + ) + .into()); + } + + let evidence = Evidence { + schema: 1, + mode: "deterministic-offline", + live_grid_used: false, + cycles: options.cycles, + operations_per_cycle: thresholds.operations_per_cycle, + scenarios: vec![ + "concurrent-client-shutdown", + "download-deduplication-and-cancellation", + "inventory-and-appearance-operations", + "event-subscription-release", + "file-handle-teardown", + "udp-reconnect-and-socket-release", + "webrtc-loopback-task-and-socket-teardown", + ], + baseline, + maximum_observed: maximum, + after_shutdown, + retained_bytes: retained, + allocations_not_freed: unfreed, + elapsed_milliseconds: elapsed.as_millis(), + thresholds, + outcome: "pass", + }; + if let Some(path) = options.evidence.as_deref() { + write_evidence(path, &evidence)?; + } + println!("{}", serde_json::to_string_pretty(&evidence)?); + Ok(()) +}