From 6c04dc380f4883c0d9c48f4fa2a1d937dfee8bc6 Mon Sep 17 00:00:00 2001 From: Alvin Chung Date: Tue, 15 Sep 2026 13:14:25 +0800 Subject: [PATCH 01/10] fix: disable native callbacks before disposing managed state --- README.md | 10 + native/yaha_native/src/binding.rs | 140 +++++++++--- native/yaha_native/src/callback_gate.rs | 139 ++++++++++++ native/yaha_native/src/context.rs | 58 ++++- native/yaha_native/src/lib.rs | 1 + .../NativeHttpHandlerCore.cs | 136 ++++++++++-- .../NativeMethods.Uwp.g.cs | 6 + src/YetAnotherHttpHandler/NativeMethods.g.cs | 6 + src/YetAnotherHttpHandler/RequestContext.cs | 83 ++++--- src/YetAnotherHttpHandler/ResponseContext.cs | 4 + .../YetAnotherHttpHandler.cs | 37 +++- .../YetAnotherHttpHandler.Test/DisposeTest.cs | 203 ++++++++++++++++++ 12 files changed, 747 insertions(+), 76 deletions(-) create mode 100644 native/yaha_native/src/callback_gate.rs create mode 100644 test/YetAnotherHttpHandler.Test/DisposeTest.cs diff --git a/README.md b/README.md index e5c4c1b..1ef044d 100644 --- a/README.md +++ b/README.md @@ -148,6 +148,14 @@ With these steps, your HttpClient is now compatible with HTTP/2.✨ However, since it does not have features such as connection control by the number of streams, it is necessary to separate handler instances when explicitly creating different connections. +### Disposing a handler + +`YetAnotherHttpHandler.Dispose()` rejects new requests and closes a native callback gate shared by response headers, response bodies, completion notifications and custom TLS certificate verification. It cancels pending requests and waits for admitted callbacks and outstanding body-flush acknowledgements before releasing their managed state. Repeated and concurrent disposal waits for the same cleanup. + +After `Dispose()` returns, this handler cannot invoke another managed callback. Native connection cleanup and OS DNS work may finish later; their native state remains alive independently. Other handlers sharing the runtime continue running. Disposal does not wait for all native work or arbitrary application continuations and `HttpContent` producers; it does stop the package's own upload loop. + +Dispose handlers **before** Unity/IL2CPP runtime cleanup or unload begins. The package does not automatically subscribe to application lifecycle events. Disposal can still block on an executing callback. Calling `Dispose()` synchronously inside a native callback, including `OnVerifyServerCertificate`, throws `InvalidOperationException` to prevent waiting for itself. A callback must also not wait for another thread to dispose the handler. + ### Using gRPC (grpc-dotnet) library To use grpc-dotnet (Grpc.Net.Client), add the following additional libraries: @@ -351,6 +359,8 @@ When creating a package, The following artifacts directory is used. - native/artifacts/{.NET RID}/{lib}yaha_native.{dll,so} +Managed sources and native binaries must come from the same build. The callback gate requires `yaha_context_disable_callbacks`, `yaha_context_wait_callbacks` and a matching response-completion handle layout. Before distributing a Unity package, update the prebuilt libraries for every platform through the `Build Native Libraries` workflow with `update-unity-native` enabled. + ```bash # Generate THIRD-PARTY-NOTICES using cargo-about cargo about generate about.hbs > ../THIRD-PARTY-NOTICES diff --git a/native/yaha_native/src/binding.rs b/native/yaha_native/src/binding.rs index ae3845f..8faaf32 100644 --- a/native/yaha_native/src/binding.rs +++ b/native/yaha_native/src/binding.rs @@ -23,6 +23,12 @@ use crate::{ primitives::WriteResult, }; use futures_util::StreamExt; +use crate::callback_gate::CallbackGuard; + +struct ResponseCompletion { + sender: oneshot::Sender>, + _guard: CallbackGuard, +} #[no_mangle] @@ -57,6 +63,9 @@ pub extern "C" fn yaha_init_runtime(worker_threads: i32) -> *mut YahaNativeRunti #[no_mangle] pub extern "C" fn yaha_dispose_runtime(ctx: *mut YahaNativeRuntimeContext) { let ctx = unsafe { Box::from_raw(ctx as *mut YahaNativeRuntimeContextInternal) }; + // Callback gates are already closed and drained. Native-only work (such + // as OS DNS resolution) can finish without blocking managed shutdown. + ctx.runtime.shutdown_background(); } #[no_mangle] @@ -72,25 +81,34 @@ pub extern "C" fn yaha_init_context( on_complete: extern "C" fn(req_seq: i32, state: NonZeroIsize, reason: CompletionReason, h2_error_code: u32), ) -> *mut YahaNativeContext { let runtime_ctx = YahaNativeRuntimeContextInternal::from_raw_context(runtime_ctx); - let ctx = Box::new(YahaNativeContextInternal::new( + let ctx = Arc::new(YahaNativeContextInternal::new( runtime_ctx.runtime.handle().clone(), on_status_code_and_headers_receive, on_receive, on_complete, )); - Box::into_raw(ctx) as *mut YahaNativeContext + Arc::into_raw(ctx) as *mut YahaNativeContext } #[no_mangle] pub extern "C" fn yaha_dispose_context(ctx: *mut YahaNativeContext) { - let mut ctx = unsafe { Box::from_raw(ctx as *mut YahaNativeContextInternal) }; - ctx.on_complete = _sentinel_on_complete; - ctx.on_receive = _sentinel_on_receive; - ctx.on_status_code_and_headers_receive = _sentinel_on_status_code_and_headers_receive; + // Requests own an Arc as well, so native cleanup can outlive Dispose. + let ctx = unsafe { Arc::from_raw(ctx as *const YahaNativeContextInternal) }; + ctx.callbacks.close(); + ctx.callbacks.wait(); +} + +#[no_mangle] +pub extern "C" fn yaha_context_disable_callbacks(ctx: *const YahaNativeContext) { + let ctx = unsafe { &*(ctx as *const YahaNativeContextInternal) }; + ctx.callbacks.close(); +} + +#[no_mangle] +pub extern "C" fn yaha_context_wait_callbacks(ctx: *const YahaNativeContext) { + let ctx = unsafe { &*(ctx as *const YahaNativeContextInternal) }; + ctx.callbacks.wait(); } -extern "C" fn _sentinel_on_complete(_: i32, _: NonZeroIsize, _: CompletionReason, _: u32) { panic!("The context has already disposed: on_complete"); } -extern "C" fn _sentinel_on_receive(_: i32, _: NonZeroIsize, _: usize, _: *const u8, _: usize) { panic!("The context has already disposed: on_receive"); } -extern "C" fn _sentinel_on_status_code_and_headers_receive(_: i32, _: NonZeroIsize, _: i32, _: YahaHttpVersion) { panic!("The context has already disposed: on_status_code_and_headers_receive"); } #[no_mangle] pub extern "C" fn yaha_client_config_add_root_certificates( @@ -486,7 +504,13 @@ pub extern "C" fn yaha_request_begin( req_ctx: *const YahaNativeRequestContext, state: NonZeroIsize ) -> bool { - let ctx = YahaNativeContextInternal::from_raw_context(ctx); + // The configured context is immutable. Keep it alive independently of + // managed SafeHandles until this request future has been dropped. + let ctx = unsafe { + let ptr = ctx as *const YahaNativeContextInternal; + Arc::increment_strong_count(ptr); + Arc::from_raw(ptr) + }; // Begin request on async runtime. let body; @@ -508,6 +532,7 @@ pub extern "C" fn yaha_request_begin( { let req_ctx = req_ctx.clone(); ctx.runtime.clone().spawn(async move { + let ctx = &*ctx; let cancellation_token = { let req_ctx = req_ctx.lock().unwrap(); req_ctx.cancellation_token.clone() @@ -532,14 +557,14 @@ pub extern "C" fn yaha_request_begin( let mut req_ctx = req_ctx.lock().unwrap(); req_ctx.last_error = Some("The client has not been built. You need to build it before sending the request.".to_string()); } - (ctx.on_complete)(seq, state, CompletionReason::Error, 0); + ctx.notify_complete(seq, state, CompletionReason::Error, 0); return; } // Send a request and wait for response status and headers. let res = select! { _ = cancellation_token.cancelled() => { - (ctx.on_complete)(seq, state, CompletionReason::Aborted, 0); + ctx.notify_complete(seq, state, CompletionReason::Aborted, 0); return; } res = ctx.request(req) => { @@ -570,7 +595,7 @@ pub extern "C" fn yaha_request_begin( req_ctx.response_status = res.status(); req_ctx.response_version = YahaHttpVersion::from(res.version()); } - (ctx.on_status_code_and_headers_receive)( + ctx.notify_headers( seq, state, res.status().as_u16() as i32, @@ -585,7 +610,7 @@ pub extern "C" fn yaha_request_begin( while !body.is_end_stream() { select! { _ = cancellation_token.cancelled() => { - (ctx.on_complete)(seq, state, CompletionReason::Aborted, 0); + ctx.notify_complete(seq, state, CompletionReason::Aborted, 0); return; } received = body.frame() => { @@ -595,10 +620,17 @@ pub extern "C" fn yaha_request_begin( Ok(frame) => { if frame.is_data() { let data = frame.into_data().unwrap(); + let Some(guard) = ctx.callbacks.enter() else { + return; + }; let (tx, rx) = oneshot::channel::>(); - let tx = Box::into_raw(Box::new(tx)) as usize; + let tx = Box::into_raw(Box::new(ResponseCompletion { + sender: tx, + _guard: guard.defer(), + })) as usize; (ctx.on_receive)(seq, state, data.len(), data.as_ptr(), tx); + drop(guard); match rx.await { Ok(result) => { if let Err(err) = result { @@ -607,7 +639,7 @@ pub extern "C" fn yaha_request_begin( let mut req_ctx = req_ctx.lock().unwrap(); req_ctx.last_error = Some(err); } - (ctx.on_complete)(seq, state, CompletionReason::Error, 0); + ctx.notify_complete(seq, state, CompletionReason::Error, 0); return; } }, @@ -617,7 +649,7 @@ pub extern "C" fn yaha_request_begin( let mut req_ctx = req_ctx.lock().unwrap(); req_ctx.last_error = Some("on_receive() has not completed correctly.".to_string()); } - (ctx.on_complete)(seq, state, CompletionReason::Error, 0); + ctx.notify_complete(seq, state, CompletionReason::Error, 0); return; } } @@ -659,7 +691,7 @@ pub extern "C" fn yaha_request_begin( let rc = reason.map(|r| u32::from(r)); - (ctx.on_complete)(seq, state, CompletionReason::Error, rc.unwrap_or_default()); + ctx.notify_complete(seq, state, CompletionReason::Error, rc.unwrap_or_default()); return; } } @@ -678,7 +710,7 @@ pub extern "C" fn yaha_request_begin( req_ctx.try_complete(); } - (ctx.on_complete)(seq, state, CompletionReason::Success, 0); + ctx.notify_complete(seq, state, CompletionReason::Success, 0); { let mut req_ctx = req_ctx.lock().unwrap(); @@ -691,7 +723,7 @@ pub extern "C" fn yaha_request_begin( true } -fn complete_with_error(ctx: &mut YahaNativeContextInternal, req_ctx: Arc>, seq: i32, state: NonZeroIsize, err: hyper_util::client::legacy::Error) { +fn complete_with_error(ctx: &YahaNativeContextInternal, req_ctx: Arc>, seq: i32, state: NonZeroIsize, err: hyper_util::client::legacy::Error) { let mut h2_error_code = None; { @@ -710,7 +742,7 @@ fn complete_with_error(ctx: &mut YahaNativeContextInternal, req_ctx: Arc>) }; + let completion = unsafe { Box::from_raw(task_handle as *mut ResponseCompletion) }; + let ResponseCompletion { sender: tx, _guard } = *completion; if error.is_null() { - tx.send(Ok(())).unwrap(); + let _ = tx.send(Ok(())); } else { let error = unsafe { (*error).to_str().to_string() }; - tx.send(Err(error)).unwrap(); + let _ = tx.send(Err(error)); } } #[cfg(test)] mod tests { + use super::*; use std::ptr::null; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::mpsc; - use super::*; + extern "C" fn headers(_: i32, state: NonZeroIsize, _: i32, _: YahaHttpVersion) { + unsafe { &*(state.get() as *const AtomicUsize) }.fetch_add(1, Ordering::SeqCst); + } + + extern "C" fn complete(_: i32, state: NonZeroIsize, _: CompletionReason, _: u32) { + unsafe { &*(state.get() as *const AtomicUsize) }.fetch_add(1, Ordering::SeqCst); + } + + extern "C" fn receive(_: i32, _: NonZeroIsize, _: usize, _: *const u8, _: usize) {} + + #[test] + fn retained_context_suppresses_late_callbacks_after_managed_owner_is_released() { + let calls = AtomicUsize::new(0); + let state = NonZeroIsize::new(&calls as *const _ as isize).unwrap(); + let runtime = yaha_init_runtime(1); + let ctx = yaha_init_context(runtime, headers, receive, complete); + let retained = unsafe { + let ptr = ctx as *const YahaNativeContextInternal; + Arc::increment_strong_count(ptr); + Arc::from_raw(ptr) + }; + retained.notify_headers(1, state, 200, YahaHttpVersion::Http11); + retained.notify_complete(1, state, CompletionReason::Success, 0); + assert_eq!(calls.load(Ordering::SeqCst), 2); + + yaha_context_disable_callbacks(ctx); + yaha_context_wait_callbacks(ctx); + yaha_dispose_context(ctx); + yaha_dispose_runtime(runtime); + + // Simulate a request which still owns the context after Dispose. + retained.notify_headers(1, state, 200, YahaHttpVersion::Http11); + retained.notify_complete(1, state, CompletionReason::Aborted, 0); + assert_eq!(calls.load(Ordering::SeqCst), 2); + } + + #[test] + fn runtime_release_does_not_join_native_blocking_work() { + let runtime = yaha_init_runtime(1); + let handle = YahaNativeRuntimeContextInternal::from_raw_context(runtime).runtime.handle().clone(); + let (entered_tx, entered_rx) = mpsc::channel(); + let (release_tx, release_rx) = mpsc::channel(); + let (done_tx, done_rx) = mpsc::channel(); + handle.spawn_blocking(move || { + entered_tx.send(()).unwrap(); + let released = release_rx.recv_timeout(Duration::from_secs(5)).is_ok(); + done_tx.send(released).unwrap(); + }); + entered_rx.recv_timeout(Duration::from_secs(5)).unwrap(); + yaha_dispose_runtime(runtime); + // A blocking Runtime drop would wait for the timeout and fail this test. + assert!(matches!(done_rx.try_recv(), Err(mpsc::TryRecvError::Empty))); + release_tx.send(()).unwrap(); + assert!(done_rx.recv_timeout(Duration::from_secs(5)).unwrap()); + } unsafe fn set_header(req_ctx: *const YahaNativeRequestContext, name: &[u8], value: &[u8]) -> bool { let name = StringBuffer::new(name.as_ptr(), name.len() as i32); diff --git a/native/yaha_native/src/callback_gate.rs b/native/yaha_native/src/callback_gate.rs new file mode 100644 index 0000000..57e4e8e --- /dev/null +++ b/native/yaha_native/src/callback_gate.rs @@ -0,0 +1,139 @@ +use std::sync::{Arc, Condvar, Mutex}; + +/// Admission is checked and counted under the same lock. Closing the gate +/// prevents even entry into the managed reverse-P/Invoke wrapper. +#[derive(Clone, Debug, Default)] +pub struct CallbackGate(Arc); + +#[derive(Debug, Default)] +struct Inner { + state: Mutex, + idle: Condvar, +} + +#[derive(Debug, Default)] +struct State { + closed: bool, + active: usize, +} + +pub struct CallbackGuard(Arc); + +impl CallbackGate { + pub fn enter(&self) -> Option { + let mut state = self.0.state.lock().unwrap(); + if state.closed { + return None; + } + state.active += 1; + Some(CallbackGuard(self.0.clone())) + } + + /// Separate closing from waiting so managed code can unblock pending + /// response flushes before waiting for their acknowledgements. + pub fn close(&self) { + self.0.state.lock().unwrap().closed = true; + } + + pub fn wait(&self) { + let mut state = self.0.state.lock().unwrap(); + assert!(state.closed); + while state.active != 0 { + state = self.0.idle.wait(state).unwrap(); + } + } +} + +impl CallbackGuard { + /// Extend an already admitted callback to cover a deferred managed body + /// acknowledgement. This must remain possible if close races the callback. + pub fn defer(&self) -> Self { + self.0.state.lock().unwrap().active += 1; + Self(self.0.clone()) + } +} + +impl Drop for CallbackGuard { + fn drop(&mut self) { + let mut state = self.0.state.lock().unwrap(); + state.active -= 1; + if state.active == 0 { + self.0.idle.notify_all(); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::{sync::mpsc, thread, time::Duration}; + + const TIMEOUT: Duration = Duration::from_secs(5); + + #[test] + fn close_rejects_new_callbacks_and_waits_for_an_admitted_callback() { + let gate = CallbackGate::default(); + let guard = gate.enter().unwrap(); + gate.close(); + assert!(gate.enter().is_none()); + let stopping = gate.clone(); + let (done_tx, done_rx) = mpsc::channel(); + let waiter = thread::spawn(move || { + stopping.wait(); + done_tx.send(()).unwrap(); + }); + assert!(matches!( + done_rx.recv_timeout(Duration::from_millis(50)), + Err(mpsc::RecvTimeoutError::Timeout) + )); + drop(guard); + done_rx.recv_timeout(TIMEOUT).unwrap(); + waiter.join().unwrap(); + gate.close(); + gate.wait(); + } + + #[test] + fn deferred_acknowledgement_outlives_callback_even_when_close_races_it() { + let gate = CallbackGate::default(); + let guard = gate.enter().unwrap(); + gate.close(); + let acknowledgement = guard.defer(); + drop(guard); + let stopping = gate.clone(); + let (done_tx, done_rx) = mpsc::channel(); + let waiter = thread::spawn(move || { + stopping.wait(); + done_tx.send(()).unwrap(); + }); + assert!(matches!( + done_rx.recv_timeout(Duration::from_millis(50)), + Err(mpsc::RecvTimeoutError::Timeout) + )); + drop(acknowledgement); + done_rx.recv_timeout(TIMEOUT).unwrap(); + waiter.join().unwrap(); + } + + #[test] + fn closing_one_handler_does_not_close_another_or_join_native_work() { + let runtime = tokio::runtime::Runtime::new().unwrap(); + let gate = CallbackGate::default(); + let other = CallbackGate::default(); + let late_callback = gate.clone(); + let (entered_tx, entered_rx) = mpsc::channel(); + let (release_tx, release_rx) = mpsc::channel(); + let work = runtime.spawn_blocking(move || { + entered_tx.send(()).unwrap(); + release_rx.recv_timeout(TIMEOUT).unwrap(); + assert!(late_callback.enter().is_none()); + }); + entered_rx.recv_timeout(TIMEOUT).unwrap(); + gate.close(); + gate.wait(); + assert!(!work.is_finished()); + assert!(other.enter().is_some()); + release_tx.send(()).unwrap(); + runtime.block_on(work).unwrap(); + } +} diff --git a/native/yaha_native/src/context.rs b/native/yaha_native/src/context.rs index 446e689..afab150 100644 --- a/native/yaha_native/src/context.rs +++ b/native/yaha_native/src/context.rs @@ -29,6 +29,8 @@ use hyperlocal::UnixConnector; use rustls::pki_types::{CertificateDer, PrivateKeyDer}; use tokio_util::sync::CancellationToken; +use crate::callback_gate::CallbackGate; + use crate::{primitives::{CompletionReason, YahaHttpVersion}}; type OnStatusCodeAndHeadersReceive = @@ -64,6 +66,7 @@ impl YahaNativeRuntimeContextInternal { pub struct YahaNativeContext; pub struct YahaNativeContextInternal<'a> { pub runtime: tokio::runtime::Handle, + pub callbacks: CallbackGate, pub client_builder: Option, pub skip_certificate_verification: Option, pub server_certificate_verification_handler: Option<(OnServerCertificateVerificationHandler, NonZeroIsize)>, @@ -96,6 +99,7 @@ impl YahaNativeContextInternal<'_> { ) -> Self { YahaNativeContextInternal { runtime: runtime_handle, + callbacks: CallbackGate::default(), tcp_client: None, client_builder: Some(Client::builder(TokioExecutor::new())), skip_certificate_verification: None, @@ -115,6 +119,18 @@ impl YahaNativeContextInternal<'_> { } } + pub fn notify_headers(&self, seq: i32, state: NonZeroIsize, status: i32, version: YahaHttpVersion) { + if let Some(_guard) = self.callbacks.enter() { + (self.on_status_code_and_headers_receive)(seq, state, status, version); + } + } + + pub fn notify_complete(&self, seq: i32, state: NonZeroIsize, reason: CompletionReason, h2_error_code: u32) { + if let Some(_guard) = self.callbacks.enter() { + (self.on_complete)(seq, state, reason, h2_error_code); + } + } + pub fn build_client(&mut self) { let mut builder = self.client_builder.take().unwrap(); builder.timer(TokioTimer::new()); @@ -151,6 +167,7 @@ impl YahaNativeContextInternal<'_> { danger::CustomCerficateVerification::new( server_certificate_verification_handler, signature_algorithms, + self.callbacks.clone(), ), )) .with_no_client_auth(); @@ -256,6 +273,7 @@ mod danger { use rustls::pki_types::{CertificateDer, ServerName, UnixTime}; use super::OnServerCertificateVerificationHandler; + use crate::callback_gate::CallbackGate; #[derive(Debug)] pub struct NoCertificateVerification {} @@ -263,16 +281,19 @@ mod danger { pub struct CustomCerficateVerification { handler: (OnServerCertificateVerificationHandler, NonZeroIsize), signature_algorithms: WebPkiSupportedAlgorithms, + callbacks: CallbackGate, } impl CustomCerficateVerification { pub fn new( handler: (OnServerCertificateVerificationHandler, NonZeroIsize), signature_algorithms: WebPkiSupportedAlgorithms, + callbacks: CallbackGate, ) -> Self { Self { handler, signature_algorithms, + callbacks, } } } @@ -308,6 +329,9 @@ mod danger { _ocsp_response: &[u8], now: UnixTime, ) -> Result { + let Some(_guard) = self.callbacks.enter() else { + return Err(Error::InvalidCertificate(rustls::CertificateError::ApplicationVerificationFailure)); + }; let server_name = server_name.to_str(); let server_name = server_name.as_bytes(); let cetificate_der = end_entity.as_ref(); @@ -434,17 +458,22 @@ mod tests { io::{self, Cursor}, net::{TcpListener, TcpStream}, num::NonZeroIsize, - sync::Arc, + sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, + }, thread, time::Duration, }; use rustls::{ - pki_types::{CertificateDer, PrivateKeyDer, ServerName}, + client::danger::ServerCertVerifier, + pki_types::{CertificateDer, PrivateKeyDer, ServerName, UnixTime}, ClientConfig, ClientConnection, ServerConfig, ServerConnection, SupportedProtocolVersion, }; use super::danger::CustomCerficateVerification; + use crate::callback_gate::CallbackGate; const CERTIFICATE_A: &[u8] = include_bytes!("../testdata/certificate-verify/cert-a.pem"); const PRIVATE_KEY_A: &[u8] = include_bytes!("../testdata/certificate-verify/key-a.pem"); @@ -515,6 +544,7 @@ mod tests { NonZeroIsize::new(1).expect("the callback state must be non-zero"), ), rustls::crypto::ring::default_provider().signature_verification_algorithms, + CallbackGate::default(), ); let client_config = ClientConfig::builder_with_protocol_versions(&[protocol_version]) .dangerous() @@ -566,4 +596,28 @@ mod tests { complete_handshake(&rustls::version::TLS13, PRIVATE_KEY_B) .expect_err("a TLS 1.3 signature made by an unrelated key must be rejected"); } + + extern "C" fn verify(state: NonZeroIsize, _: *const u8, _: usize, _: *const u8, _: usize, _: u64) -> bool { + unsafe { &*(state.get() as *const AtomicUsize) }.fetch_add(1, Ordering::SeqCst); + true + } + + #[test] + fn previously_created_tls_verifier_observes_closed_gate() { + let calls = AtomicUsize::new(0); + let callbacks = CallbackGate::default(); + let verifier = CustomCerficateVerification::new( + (verify, NonZeroIsize::new(&calls as *const _ as isize).unwrap()), + rustls::crypto::ring::default_provider().signature_verification_algorithms, + callbacks.clone(), + ); + let certificate = CertificateDer::from(&b"test certificate"[..]); + let name = ServerName::try_from("localhost").unwrap(); + let now = UnixTime::since_unix_epoch(Duration::from_secs(0)); + assert!(verifier.verify_server_cert(&certificate, &[], &name, &[], now).is_ok()); + callbacks.close(); + callbacks.wait(); + assert!(verifier.verify_server_cert(&certificate, &[], &name, &[], now).is_err()); + assert_eq!(calls.load(Ordering::SeqCst), 1); + } } diff --git a/native/yaha_native/src/lib.rs b/native/yaha_native/src/lib.rs index 29a702f..b6a3f53 100644 --- a/native/yaha_native/src/lib.rs +++ b/native/yaha_native/src/lib.rs @@ -1,4 +1,5 @@ mod context; +mod callback_gate; mod interop; mod primitives; diff --git a/src/YetAnotherHttpHandler/NativeHttpHandlerCore.cs b/src/YetAnotherHttpHandler/NativeHttpHandlerCore.cs index c32c04f..066d654 100644 --- a/src/YetAnotherHttpHandler/NativeHttpHandlerCore.cs +++ b/src/YetAnotherHttpHandler/NativeHttpHandlerCore.cs @@ -1,4 +1,5 @@ using System; +using System.Collections.Concurrent; using System.Diagnostics; using System.Text; using System.IO.Pipelines; @@ -22,8 +23,25 @@ internal class NativeHttpHandlerCore : IDisposable //private unsafe YahaNativeContext* _ctx; private readonly YahaContextSafeHandle _handle; private bool _disposed = false; + private readonly object _disposeLock = new object(); + private readonly ConcurrentDictionary _requests = new ConcurrentDictionary(); private PipeOptions? _responsePipeOptions; + [ThreadStatic] + private static int _callbackDepth; + + internal static void ThrowIfInCallback() + { + if (_callbackDepth != 0) + { + throw new InvalidOperationException("Cannot synchronously dispose a handler from a native callback. Dispose it after the callback returns."); + } + } + + internal void RemoveRequest(RequestContext request) + { + _requests.TryRemove(request.RequestSequence, out _); + } // NOTE: We need to keep the callback delegates in advance. // The delegates are kept on the Rust side, so it will crash if they are garbage collected. private static readonly unsafe NativeMethods.yaha_init_context_on_status_code_and_headers_receive_delegate OnStatusCodeAndHeaderReceiveCallback = OnStatusCodeAndHeaderReceive; @@ -50,7 +68,7 @@ public unsafe NativeHttpHandlerCore(NativeClientSettings settings) catch { // NOTE: If the initialization fails, we need to release the runtime. - NativeRuntime.Instance.Release(); + Dispose(); throw; } finally @@ -237,8 +255,19 @@ public Task SendAsync(HttpRequestMessage request, Cancellat static async Task SendBodyAsync(HttpContent requestContent, PipeWriter writer, CancellationToken cancellationToken) { - await requestContent.CopyToAsync(writer.AsStream()).ConfigureAwait(false); // TODO: cancel - await writer.CompleteAsync().ConfigureAwait(false); + Exception? error = null; + try + { + await requestContent.CopyToAsync(writer.AsStream()).ConfigureAwait(false); // TODO: cancel + } + catch (Exception ex) + { + error = ex; + } + finally + { + await writer.CompleteAsync(error).ConfigureAwait(false); + } } } @@ -271,6 +300,14 @@ private unsafe RequestContext Send(HttpRequestMessage request, CancellationToken return UnsafeSend(_handle, reqCtxHandle, requestSequence, request, cancellationToken); } + catch + { + // Header/URI validation can fail before a managed request + // is registered. Do not leave its parent/runtime references + // waiting for a SafeHandle finalizer after handler disposal. + reqCtxHandle.Dispose(); + throw; + } finally { if (addRefReqContext) @@ -340,8 +377,9 @@ private unsafe RequestContext UnsafeSend(YahaContextSafeHandle ctxHandle, YahaRe NativeMethods.yaha_request_set_has_body(ctx, reqCtx, request.Content != null); // Prepare a request context - var requestContextManaged = new RequestContext(_handle, reqCtxHandle, request, requestSequence, _responsePipeOptions, cancellationToken); + var requestContextManaged = new RequestContext(this, _handle, reqCtxHandle, request, requestSequence, _responsePipeOptions, cancellationToken); requestContextManaged.Allocate(); + _requests.TryAdd(requestSequence, requestContextManaged); if (cancellationToken.IsCancellationRequested) { // Dispose the request context immediately. @@ -385,6 +423,19 @@ private static unsafe void SetRequestHeaders(HttpHeaders headers, YahaNativeCont [MonoPInvokeCallback(typeof(NativeMethods.yaha_init_context_on_status_code_and_headers_receive_delegate))] private static unsafe void OnStatusCodeAndHeaderReceive(int reqSeq, IntPtr state, int statusCode, YahaHttpVersion version) + { + _callbackDepth++; + try + { + _OnStatusCodeAndHeaderReceive(reqSeq, state, statusCode, version); + } + finally + { + _callbackDepth--; + } + } + + private static unsafe void _OnStatusCodeAndHeaderReceive(int reqSeq, IntPtr state, int statusCode, YahaHttpVersion version) { if (YahaEventSource.Log.IsEnabled()) YahaEventSource.Log.Info($"[ReqSeq:{reqSeq}:State:0x{state:X}] Status code and headers received: StatusCode={statusCode}; Version={version}"); @@ -437,6 +488,19 @@ private static unsafe void OnStatusCodeAndHeaderReceive(int reqSeq, IntPtr state [MonoPInvokeCallback(typeof(NativeMethods.yaha_client_config_set_server_certificate_verification_handler_handler_delegate))] private static unsafe bool OnServerCertificateVerification(IntPtr callbackState, byte* serverNamePtr, UIntPtr /*nuint*/ serverNameLength, byte* certificateDerPtr, UIntPtr /*nuint*/ certificateDerLength, ulong now) + { + _callbackDepth++; + try + { + return _OnServerCertificateVerification(callbackState, serverNamePtr, serverNameLength, certificateDerPtr, certificateDerLength, now); + } + finally + { + _callbackDepth--; + } + } + + private static unsafe bool _OnServerCertificateVerification(IntPtr callbackState, byte* serverNamePtr, UIntPtr /*nuint*/ serverNameLength, byte* certificateDerPtr, UIntPtr /*nuint*/ certificateDerLength, ulong now) { var serverName = UnsafeUtilities.GetStringFromUtf8Bytes(new ReadOnlySpan(serverNamePtr, (int)serverNameLength)); var certificateDer = new ReadOnlySpan(certificateDerPtr, (int)certificateDerLength); @@ -464,6 +528,19 @@ private static unsafe bool OnServerCertificateVerification(IntPtr callbackState, [MonoPInvokeCallback(typeof(NativeMethods.yaha_init_context_on_receive_delegate))] private static unsafe void OnReceive(int reqSeq, IntPtr state, UIntPtr length, byte* buf, nuint taskHandle) + { + _callbackDepth++; + try + { + _OnReceive(reqSeq, state, length, buf, taskHandle); + } + finally + { + _callbackDepth--; + } + } + + private static unsafe void _OnReceive(int reqSeq, IntPtr state, UIntPtr length, byte* buf, nuint taskHandle) { try { @@ -524,6 +601,19 @@ private static unsafe void CompleteTask(nuint taskHandle, string? error = null) [MonoPInvokeCallback(typeof(NativeMethods.yaha_init_context_on_complete_delegate))] private static unsafe void OnComplete(int reqSeq, IntPtr state, CompletionReason reason, uint h2ErrorCode) + { + _callbackDepth++; + try + { + _OnComplete(reqSeq, state, reason, h2ErrorCode); + } + finally + { + _callbackDepth--; + } + } + + private static unsafe void _OnComplete(int reqSeq, IntPtr state, CompletionReason reason, uint h2ErrorCode) { if (YahaEventSource.Log.IsEnabled()) YahaEventSource.Log.Info($"[ReqSeq:{reqSeq}:State:0x{state:X}] Response completed: Reason={reason}; H2ErrorCode=0x{h2ErrorCode:x}"); @@ -630,23 +720,41 @@ public void Dispose() GC.SuppressFinalize(this); } - private void Dispose(bool disposing) + private unsafe void Dispose(bool disposing) { - if (_disposed) + ThrowIfInCallback(); + lock (_disposeLock) { - return; - } + if (_disposed) + { + return; + } - if (YahaEventSource.Log.IsEnabled()) YahaEventSource.Log.Info($"Dispose {nameof(NativeHttpHandlerCore)}; disposing={disposing}"); + // Close native callback admission before cancelling managed pipes. + // Existing callbacks and deferred flush acknowledgements are + // drained while their managed state is still alive. + NativeMethods.yaha_context_disable_callbacks(_handle.DangerousGet()); + var requests = _requests.Values; + foreach (var request in requests) + { + request.Response.Cancel(); + } - NativeRuntime.Instance.Release(); // We always need to release runtime. + NativeMethods.yaha_context_wait_callbacks(_handle.DangerousGet()); + + // Disabled callbacks no longer deliver OnComplete. Release + // their state here, and finish cleanup synchronously instead of + // relying on queued ThreadPool work during runtime shutdown. + foreach (var request in requests) + { + request.Release(); + request.Dispose(); + } - if (disposing) - { _handle.Dispose(); + NativeRuntime.Instance.Release(); + _disposed = true; } - - _disposed = true; } } } diff --git a/src/YetAnotherHttpHandler/NativeMethods.Uwp.g.cs b/src/YetAnotherHttpHandler/NativeMethods.Uwp.g.cs index f6e1a33..cbdd2d1 100644 --- a/src/YetAnotherHttpHandler/NativeMethods.Uwp.g.cs +++ b/src/YetAnotherHttpHandler/NativeMethods.Uwp.g.cs @@ -44,6 +44,12 @@ internal static unsafe partial class NativeMethods [DllImport(__DllName, EntryPoint = "yaha_dispose_context", CallingConvention = CallingConvention.Cdecl, ExactSpelling = true)] public static extern void yaha_dispose_context(YahaNativeContext* ctx); + [DllImport(__DllName, EntryPoint = "yaha_context_disable_callbacks", CallingConvention = CallingConvention.Cdecl, ExactSpelling = true)] + public static extern void yaha_context_disable_callbacks(YahaNativeContext* ctx); + + [DllImport(__DllName, EntryPoint = "yaha_context_wait_callbacks", CallingConvention = CallingConvention.Cdecl, ExactSpelling = true)] + public static extern void yaha_context_wait_callbacks(YahaNativeContext* ctx); + [DllImport(__DllName, EntryPoint = "yaha_client_config_add_root_certificates", CallingConvention = CallingConvention.Cdecl, ExactSpelling = true)] public static extern nuint yaha_client_config_add_root_certificates(YahaNativeContext* ctx, StringBuffer* root_certs); diff --git a/src/YetAnotherHttpHandler/NativeMethods.g.cs b/src/YetAnotherHttpHandler/NativeMethods.g.cs index b060637..bbf24c9 100644 --- a/src/YetAnotherHttpHandler/NativeMethods.g.cs +++ b/src/YetAnotherHttpHandler/NativeMethods.g.cs @@ -49,6 +49,12 @@ internal static unsafe partial class NativeMethods [DllImport(__DllName, EntryPoint = "yaha_dispose_context", CallingConvention = CallingConvention.Cdecl, ExactSpelling = true)] public static extern void yaha_dispose_context(YahaNativeContext* ctx); + [DllImport(__DllName, EntryPoint = "yaha_context_disable_callbacks", CallingConvention = CallingConvention.Cdecl, ExactSpelling = true)] + public static extern void yaha_context_disable_callbacks(YahaNativeContext* ctx); + + [DllImport(__DllName, EntryPoint = "yaha_context_wait_callbacks", CallingConvention = CallingConvention.Cdecl, ExactSpelling = true)] + public static extern void yaha_context_wait_callbacks(YahaNativeContext* ctx); + [DllImport(__DllName, EntryPoint = "yaha_client_config_add_root_certificates", CallingConvention = CallingConvention.Cdecl, ExactSpelling = true)] public static extern nuint yaha_client_config_add_root_certificates(YahaNativeContext* ctx, StringBuffer* root_certs); diff --git a/src/YetAnotherHttpHandler/RequestContext.cs b/src/YetAnotherHttpHandler/RequestContext.cs index e2292be..7a6edea 100644 --- a/src/YetAnotherHttpHandler/RequestContext.cs +++ b/src/YetAnotherHttpHandler/RequestContext.cs @@ -13,6 +13,9 @@ namespace Cysharp.Net.Http { internal class RequestContext : IDisposable { + private readonly NativeHttpHandlerCore _owner; + private readonly object _disposeLock = new object(); + private bool _disposed; private readonly Pipe _pipe = new(PipeOptions.Default); private readonly CancellationTokenSource _cancellationTokenSource; private readonly int _requestSequence; @@ -34,8 +37,9 @@ internal class RequestContext : IDisposable public PipeWriter Writer => _pipe.Writer; public IntPtr Handle => GCHandle.ToIntPtr(_handle); - internal RequestContext(YahaContextSafeHandle ctx, YahaRequestContextSafeHandle requestContext, HttpRequestMessage requestMessage, int requestSequence, PipeOptions? responsePipeOptions, CancellationToken cancellationToken) + internal RequestContext(NativeHttpHandlerCore owner, YahaContextSafeHandle ctx, YahaRequestContextSafeHandle requestContext, HttpRequestMessage requestMessage, int requestSequence, PipeOptions? responsePipeOptions, CancellationToken cancellationToken) { + _owner = owner; _ctxHandle = ctx; _requestContextHandle = requestContext; _response = new ResponseContext(requestMessage, this, responsePipeOptions, cancellationToken); @@ -49,14 +53,21 @@ internal RequestContext(YahaContextSafeHandle ctx, YahaRequestContextSafeHandle internal void Start(bool hasBody) { - if (hasBody) + lock (_disposeLock) { - _readRequestTask = RunReadRequestLoopAsync(_cancellationTokenSource.Token); - } - else - { - if (YahaEventSource.Log.IsEnabled()) YahaEventSource.Log.Trace($"[ReqSeq:{_requestSequence}:State:0x{Handle:X}] The request has no body. Complete immediately."); - TryCompleteBody(); + // A very fast response can finish before request_begin returns. + if (_disposed) + { + return; + } + if (hasBody) + { + _readRequestTask = RunReadRequestLoopAsync(_cancellationTokenSource.Token); + } + else + { + TryCompleteBody(); + } } } @@ -75,9 +86,14 @@ public void Allocate() /// public void Release() { - Debug.Assert(_handle.IsAllocated); lock (_handleLock) { + // Native completion and handler shutdown both own a cleanup path. + // Shutdown only calls Release after callbacks have been disabled and drained. + if (!_handle.IsAllocated) + { + return; + } if (YahaEventSource.Log.IsEnabled()) YahaEventSource.Log.Trace($"[ReqSeq:{_requestSequence}:State:0x{Handle:X}] Releasing state"); _handle.Free(); _handle = default; @@ -141,7 +157,7 @@ private void WriteBody(Span data) { // TODO: retryAfter += retryInterval; - Thread.Sleep(Math.Min(1000, retryAfter)); + _cancellationTokenSource.Token.WaitHandle.WaitOne(Math.Min(1000, retryAfter)); } } @@ -352,27 +368,36 @@ public void Dispose() private void Dispose(bool disposing) { - if (YahaEventSource.Log.IsEnabled()) YahaEventSource.Log.Info($"[ReqSeq:{_requestSequence}:State:0x{Handle:X}] Dispose RequestContext: disposing={disposing}"); - - // Abort the request and dispose the request context handle whether called from manual Dispose or the finalizer. - TryAbort(); - - if (disposing) - { - _cancellationTokenSource.Cancel(); - _cancellationTokenSource.Dispose(); - // DO NOT Dispose `_ctx` here. - } - else + lock (_disposeLock) { - // Executing within the finalizer thread. - // NOTE: Waits by blocking until the request is completed on the native side. - // If not waited here, issues such as crashes may occur when callbacks are invoked after the .NET side is destroyed by Unity's Domain Reload. - // However, caution is needed with the invocation order and timing of callbacks, as well as the handling of locks, since the finalizer thread may become blocked. + if (_disposed) + { + return; + } + + TryAbort(); _fullyCompleted.Wait(); - } + _cancellationTokenSource.Cancel(); - TryReleaseNativeHandles(); + // The upload loop also calls native methods. Join it before + // releasing handles or disposing its cancellation source. + if (_readRequestTask != null) + { + _readRequestTask.GetAwaiter().GetResult(); + } + else + { + TryCompleteBody(); + } + + TryReleaseNativeHandles(); + if (disposing) + { + _cancellationTokenSource.Dispose(); + } + _disposed = true; + _owner.RemoveRequest(this); + } } } -} \ No newline at end of file +} diff --git a/src/YetAnotherHttpHandler/ResponseContext.cs b/src/YetAnotherHttpHandler/ResponseContext.cs index c0349e3..a83482b 100644 --- a/src/YetAnotherHttpHandler/ResponseContext.cs +++ b/src/YetAnotherHttpHandler/ResponseContext.cs @@ -164,6 +164,10 @@ public void Cancel() { if (YahaEventSource.Log.IsEnabled()) YahaEventSource.Log.Trace($"[ReqSeq:{_requestContext.RequestSequence}] Response was cancelled"); + // A disposing handler can cancel a response whose caller's token + // was never cancelled. Unblock backpressure BEFORE taking the lock: + // a completion callback may hold it while waiting for this flush. + _pipe.Writer.CancelPendingFlush(); lock (_writeLock) { if (_completed) return; diff --git a/src/YetAnotherHttpHandler/YetAnotherHttpHandler.cs b/src/YetAnotherHttpHandler/YetAnotherHttpHandler.cs index 42f60b7..8c4aa58 100644 --- a/src/YetAnotherHttpHandler/YetAnotherHttpHandler.cs +++ b/src/YetAnotherHttpHandler/YetAnotherHttpHandler.cs @@ -13,6 +13,7 @@ namespace Cysharp.Net.Http public class YetAnotherHttpHandler : HttpMessageHandler { private readonly NativeClientSettings _settings = new NativeClientSettings(); + private readonly object _lifetimeLock = new object(); private bool _disposed; private NativeHttpHandlerCore? _handler; @@ -207,16 +208,40 @@ private NativeHttpHandlerCore SetupHandler() protected override Task SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) { - ThrowIfDisposed(); - var handler = _handler ?? SetupHandler(); - return handler.SendAsync(request, cancellationToken); + lock (_lifetimeLock) + { + ThrowIfDisposed(); + var handler = _handler ?? SetupHandler(); + return handler.SendAsync(request, cancellationToken); + } } + /// + /// Disables native callbacks, cancels pending requests and waits for + /// admitted callbacks to return before releasing their managed state. + /// + /// + /// Call before shutting down the managed runtime. Synchronous disposal + /// from a native callback (including certificate verification) is not supported. + /// protected override void Dispose(bool disposing) { - _handler?.Dispose(); - _handler = null; - _disposed = true; + NativeHttpHandlerCore.ThrowIfInCallback(); + NativeHttpHandlerCore? handler; + lock (_lifetimeLock) + { + _disposed = true; + handler = _handler; + } + + // Do not hold the admission lock while joining callbacks: user + // certificate verification may try to send another request, which + // must observe ObjectDisposedException instead of deadlocking. + handler?.Dispose(); + lock (_lifetimeLock) + { + _handler = null; + } } private void ThrowIfDisposed() diff --git a/test/YetAnotherHttpHandler.Test/DisposeTest.cs b/test/YetAnotherHttpHandler.Test/DisposeTest.cs new file mode 100644 index 0000000..4525163 --- /dev/null +++ b/test/YetAnotherHttpHandler.Test/DisposeTest.cs @@ -0,0 +1,203 @@ +using System.IO.Pipelines; +using Cysharp.Net.Http; + +namespace _YetAnotherHttpHandler.Test; + +[Collection(nameof(YetAnotherHttpHandlerTest))] +public class DisposeTest(ITestOutputHelper testOutputHelper) : UseTestServerTestBase(testOutputHelper) +{ + protected override TimeSpan UnexpectedTimeout => TimeSpan.FromSeconds(30); + + [Fact] + public async Task Dispose_ReleasesRequestsThatFailBeforeNativeWorkStarts() + { + using var handler = new NativeHttpHandlerCore(new NativeClientSettings()); + using var request = new HttpRequestMessage(HttpMethod.Get, new Uri("relative", UriKind.Relative)); + // Call the core directly so URI conversion fails after native request + // allocation, instead of HttpClient rejecting it before allocation. + await Assert.ThrowsAsync(() => handler.SendAsync(request, CancellationToken.None)); + + var runtime = NativeRuntime.Instance.Acquire(); + NativeRuntime.Instance.Release(); + handler.Dispose(); + + // A failed, unregistered request must not keep the runtime alive until + // GC finalizes its SafeHandle. No GC.Collect is needed for disposal. + Assert.True(runtime.IsClosed); + } + + [Fact] + public async Task Dispose_CancelsPendingHeaders_WhileAnotherHandlerKeepsWorking() + { + await using var server = await LaunchServerAsync(TestServerListenMode.InsecureHttp1Only); + using var handler = new YetAnotherHttpHandler(); + using var client = new HttpClient(handler, disposeHandler: false); + using var otherHandler = new YetAnotherHttpHandler(); + using var otherClient = new HttpClient(otherHandler, disposeHandler: false); + Assert.Equal("__OK__", await otherClient.GetStringAsync(server.BaseUri, TimeoutToken)); + + var response = client.GetAsync($"{server.BaseUri}/slow-response-headers", TimeoutToken); + await _DisposeOnDedicatedThread(handler).WaitAsync(TimeoutToken); + + // HttpClient's managed continuations can run after native work stops. + await _AssertRequestStoppedAsync(response); + Assert.Equal("__OK__", await otherClient.GetStringAsync(server.BaseUri, TimeoutToken)); + await Assert.ThrowsAsync(() => client.GetAsync(server.BaseUri, TimeoutToken)); + handler.Dispose(); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task Dispose_UnblocksUnreadResponseBody(bool http2) + { + await using var server = await LaunchServerAsync(http2 + ? TestServerListenMode.InsecureHttp2Only : TestServerListenMode.InsecureHttp1Only); + using var handler = new YetAnotherHttpHandler + { + Http2Only = http2, + ResponsePipeOptions = new PipeOptions(pauseWriterThreshold: 1, resumeWriterThreshold: 1, useSynchronizationContext: false), + }; + using var client = new HttpClient(handler, disposeHandler: false); + // No request cancellation token: Dispose itself must unblock the flush. + using var response = await client.GetAsync($"{server.BaseUri}/random?size=1048576", HttpCompletionOption.ResponseHeadersRead).WaitAsync(TimeoutToken); + using var body = await response.Content.ReadAsStreamAsync(TimeoutToken); + Assert.Equal(1, await body.ReadAsync(new byte[1], TimeoutToken)); + + await _DisposeOnDedicatedThread(handler).WaitAsync(TimeoutToken); + + await _AssertRequestStoppedAsync(body.CopyToAsync(Stream.Null, TimeoutToken)); + } + + [Fact] + public async Task Dispose_UnblocksUploadToServerThatDoesNotRead() + { + await using var server = await LaunchServerAsync(TestServerListenMode.InsecureHttp2Only); + using var handler = new YetAnotherHttpHandler + { + Http2Only = true, + }; + using var client = new HttpClient(handler, disposeHandler: false); + using var request = new HttpRequestMessage(HttpMethod.Post, $"{server.BaseUri}/post-never-read") + { + Content = new ByteArrayContent(new byte[8 * 1024 * 1024]), + }; + using var response = await client.SendAsync(request, HttpCompletionOption.ResponseHeadersRead).WaitAsync(TimeoutToken); + + await _DisposeOnDedicatedThread(handler).WaitAsync(TimeoutToken); + + await _AssertRequestStoppedAsync(response.Content.CopyToAsync(Stream.Null, TimeoutToken)); + } + + [Fact] + public async Task Dispose_WaitsForCertificateCallback_AndConcurrentDispose() + { + await using var server = await LaunchServerAsync(TestServerListenMode.SecureHttp2Only); + using var callbackRelease = new ManualResetEventSlim(); + var callbackEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var callbackReturned = false; + using var handler = new YetAnotherHttpHandler + { + OnVerifyServerCertificate = (_, _, _) => + { + callbackEntered.TrySetResult(); + callbackRelease.Wait(TimeoutToken); + callbackReturned = true; + return true; + }, + }; + using var client = new HttpClient(handler, disposeHandler: false); + var response = client.GetAsync(server.BaseUri); + Task? first = null; + Task? second = null; + try + { + await callbackEntered.Task.WaitAsync(TimeoutToken); + first = _DisposeOnDedicatedThread(handler); + // Cancellation of the pending response proves Dispose has started. + await _AssertRequestStoppedAsync(response); + Assert.False(first.IsCompleted); + second = _DisposeOnDedicatedThread(handler); + // The first Dispose is held inside the verifier. A second caller + // must not return early just because the public handler is closed. + Assert.NotSame(second, await Task.WhenAny(second, Task.Delay(100, TimeoutToken))); + await Assert.ThrowsAsync(() => client.GetAsync(server.BaseUri)); + } + finally + { + callbackRelease.Set(); + if (first != null) + { + await first.WaitAsync(TimeoutToken); + } + if (second != null) + { + await second.WaitAsync(TimeoutToken); + } + } + Assert.True(callbackReturned); + } + + [Fact] + public async Task Dispose_FromCertificateCallback_IsRejectedWithoutDisposingHandler() + { + await using var server = await LaunchServerAsync(TestServerListenMode.SecureHttp2Only); + using var handler = new YetAnotherHttpHandler(); + Exception? disposalError = null; + handler.OnVerifyServerCertificate = (_, _, _) => + { + disposalError = Record.Exception(handler.Dispose); + return true; + }; + using var client = new HttpClient(handler, disposeHandler: false); + + Assert.Equal("__OK__", await client.GetStringAsync(server.BaseUri, TimeoutToken)); + Assert.IsType(disposalError); + Assert.Equal("__OK__", await client.GetStringAsync(server.BaseUri, TimeoutToken)); + await _DisposeOnDedicatedThread(handler).WaitAsync(TimeoutToken); + } + + [Fact] + public async Task Dispose_RacesWithRequestAdmission() + { + await using var server = await LaunchServerAsync(TestServerListenMode.InsecureHttp1Only); + for (var iteration = 0; iteration < 30; iteration++) + { + using var handler = new YetAnotherHttpHandler(); + using var client = new HttpClient(handler, disposeHandler: false); + using var start = new ManualResetEventSlim(); + var sends = Enumerable.Range(0, 4).Select(_ => Task.Run(async () => + { + start.Wait(TimeoutToken); + try + { + using var response = await client.GetAsync(server.BaseUri, TimeoutToken); + response.EnsureSuccessStatusCode(); + } + catch (Exception ex) when (ex is ObjectDisposedException or HttpRequestException or OperationCanceledException) + { + // Either admitted and cancelled, or rejected at admission. + } + })).ToArray(); + var dispose = Task.Run(() => + { + start.Wait(TimeoutToken); + handler.Dispose(); + }); + start.Set(); + await Task.WhenAll(sends.Append(dispose)).WaitAsync(TimeoutToken); + } + } + + private static Task _DisposeOnDedicatedThread(YetAnotherHttpHandler handler) + { + return Task.Factory.StartNew(handler.Dispose, CancellationToken.None, TaskCreationOptions.LongRunning, TaskScheduler.Default); + } + + private static async Task _AssertRequestStoppedAsync(Task request) + { + var error = await Record.ExceptionAsync(() => request.WaitAsync(TimeSpan.FromSeconds(5))); + Assert.True(error is HttpRequestException or OperationCanceledException or IOException, + $"Expected request cancellation/failure, got {error}"); + } +} From 3a3d6df4b61507f2d1667cc87b45ca69609a3772 Mon Sep 17 00:00:00 2001 From: Alvin Chung Date: Mon, 21 Sep 2026 17:17:50 +0800 Subject: [PATCH 02/10] docs: clarify request admission guarantee during disposal --- src/YetAnotherHttpHandler/NativeHttpHandlerCore.cs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/YetAnotherHttpHandler/NativeHttpHandlerCore.cs b/src/YetAnotherHttpHandler/NativeHttpHandlerCore.cs index 066d654..76c08b9 100644 --- a/src/YetAnotherHttpHandler/NativeHttpHandlerCore.cs +++ b/src/YetAnotherHttpHandler/NativeHttpHandlerCore.cs @@ -734,6 +734,8 @@ private unsafe void Dispose(bool disposing) // Existing callbacks and deferred flush acknowledgements are // drained while their managed state is still alive. NativeMethods.yaha_context_disable_callbacks(_handle.DangerousGet()); + // The owner's admission coordination guarantees no request can + // register after this snapshot and escape cancellation and release. var requests = _requests.Values; foreach (var request in requests) { From 177534c7e2149181e9efb46322526d52bebceb09 Mon Sep 17 00:00:00 2001 From: Alvin Chung Date: Mon, 21 Sep 2026 17:31:07 +0800 Subject: [PATCH 03/10] refactor: name handler shutdown phases --- .../NativeHttpHandlerCore.cs | 54 ++++++++++--------- src/YetAnotherHttpHandler/RequestContext.cs | 1 - 2 files changed, 30 insertions(+), 25 deletions(-) diff --git a/src/YetAnotherHttpHandler/NativeHttpHandlerCore.cs b/src/YetAnotherHttpHandler/NativeHttpHandlerCore.cs index 76c08b9..29469cd 100644 --- a/src/YetAnotherHttpHandler/NativeHttpHandlerCore.cs +++ b/src/YetAnotherHttpHandler/NativeHttpHandlerCore.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Concurrent; +using System.Collections.Generic; using System.Diagnostics; using System.Text; using System.IO.Pipelines; @@ -720,7 +721,7 @@ public void Dispose() GC.SuppressFinalize(this); } - private unsafe void Dispose(bool disposing) + private void Dispose(bool disposing) { ThrowIfInCallback(); lock (_disposeLock) @@ -730,33 +731,38 @@ private unsafe void Dispose(bool disposing) return; } - // Close native callback admission before cancelling managed pipes. - // Existing callbacks and deferred flush acknowledgements are - // drained while their managed state is still alive. - NativeMethods.yaha_context_disable_callbacks(_handle.DangerousGet()); - // The owner's admission coordination guarantees no request can - // register after this snapshot and escape cancellation and release. - var requests = _requests.Values; - foreach (var request in requests) - { - request.Response.Cancel(); - } - - NativeMethods.yaha_context_wait_callbacks(_handle.DangerousGet()); - - // Disabled callbacks no longer deliver OnComplete. Release - // their state here, and finish cleanup synchronously instead of - // relying on queued ThreadPool work during runtime shutdown. - foreach (var request in requests) - { - request.Release(); - request.Dispose(); - } - + var requests = _DisableAndDrainCallbacks(); + _ReleaseDrainedRequests(requests); _handle.Dispose(); NativeRuntime.Instance.Release(); _disposed = true; } } + + private unsafe ICollection _DisableAndDrainCallbacks() + { + NativeMethods.yaha_context_disable_callbacks(_handle.DangerousGet()); + // The owner's admission coordination guarantees no request can + // register after this snapshot and escape cancellation and release. + var requests = _requests.Values; + foreach (var request in requests) + { + request.Response.Cancel(); + } + + NativeMethods.yaha_context_wait_callbacks(_handle.DangerousGet()); + return requests; + } + + private static void _ReleaseDrainedRequests(ICollection requests) + { + // Disabled callbacks no longer deliver OnComplete, and runtime + // shutdown cannot rely on queued ThreadPool work for cleanup. + foreach (var request in requests) + { + request.Release(); + request.Dispose(); + } + } } } diff --git a/src/YetAnotherHttpHandler/RequestContext.cs b/src/YetAnotherHttpHandler/RequestContext.cs index 7a6edea..3869773 100644 --- a/src/YetAnotherHttpHandler/RequestContext.cs +++ b/src/YetAnotherHttpHandler/RequestContext.cs @@ -89,7 +89,6 @@ public void Release() lock (_handleLock) { // Native completion and handler shutdown both own a cleanup path. - // Shutdown only calls Release after callbacks have been disabled and drained. if (!_handle.IsAllocated) { return; From 32c679c8c9bd43fe22bcf0583bc0b4995c8923cb Mon Sep 17 00:00:00 2001 From: Alvin Chung Date: Tue, 22 Sep 2026 11:59:57 +0800 Subject: [PATCH 04/10] refactor: consolidate native callback depth tracking --- .../NativeHttpHandlerCore.cs | 63 +++++-------------- 1 file changed, 15 insertions(+), 48 deletions(-) diff --git a/src/YetAnotherHttpHandler/NativeHttpHandlerCore.cs b/src/YetAnotherHttpHandler/NativeHttpHandlerCore.cs index 29469cd..b6f79b5 100644 --- a/src/YetAnotherHttpHandler/NativeHttpHandlerCore.cs +++ b/src/YetAnotherHttpHandler/NativeHttpHandlerCore.cs @@ -31,6 +31,17 @@ internal class NativeHttpHandlerCore : IDisposable [ThreadStatic] private static int _callbackDepth; + private readonly struct CallbackScope : IDisposable + { + public static CallbackScope Enter() + { + _callbackDepth++; + return default; + } + + public void Dispose() => _callbackDepth--; + } + internal static void ThrowIfInCallback() { if (_callbackDepth != 0) @@ -425,19 +436,8 @@ private static unsafe void SetRequestHeaders(HttpHeaders headers, YahaNativeCont [MonoPInvokeCallback(typeof(NativeMethods.yaha_init_context_on_status_code_and_headers_receive_delegate))] private static unsafe void OnStatusCodeAndHeaderReceive(int reqSeq, IntPtr state, int statusCode, YahaHttpVersion version) { - _callbackDepth++; - try - { - _OnStatusCodeAndHeaderReceive(reqSeq, state, statusCode, version); - } - finally - { - _callbackDepth--; - } - } + using var _ = CallbackScope.Enter(); - private static unsafe void _OnStatusCodeAndHeaderReceive(int reqSeq, IntPtr state, int statusCode, YahaHttpVersion version) - { if (YahaEventSource.Log.IsEnabled()) YahaEventSource.Log.Info($"[ReqSeq:{reqSeq}:State:0x{state:X}] Status code and headers received: StatusCode={statusCode}; Version={version}"); var requestContext = RequestContext.FromHandle(state); @@ -490,19 +490,8 @@ private static unsafe void _OnStatusCodeAndHeaderReceive(int reqSeq, IntPtr stat [MonoPInvokeCallback(typeof(NativeMethods.yaha_client_config_set_server_certificate_verification_handler_handler_delegate))] private static unsafe bool OnServerCertificateVerification(IntPtr callbackState, byte* serverNamePtr, UIntPtr /*nuint*/ serverNameLength, byte* certificateDerPtr, UIntPtr /*nuint*/ certificateDerLength, ulong now) { - _callbackDepth++; - try - { - return _OnServerCertificateVerification(callbackState, serverNamePtr, serverNameLength, certificateDerPtr, certificateDerLength, now); - } - finally - { - _callbackDepth--; - } - } + using var _ = CallbackScope.Enter(); - private static unsafe bool _OnServerCertificateVerification(IntPtr callbackState, byte* serverNamePtr, UIntPtr /*nuint*/ serverNameLength, byte* certificateDerPtr, UIntPtr /*nuint*/ certificateDerLength, ulong now) - { var serverName = UnsafeUtilities.GetStringFromUtf8Bytes(new ReadOnlySpan(serverNamePtr, (int)serverNameLength)); var certificateDer = new ReadOnlySpan(certificateDerPtr, (int)certificateDerLength); if (YahaEventSource.Log.IsEnabled()) YahaEventSource.Log.Trace($"OnServerCertificateVerification: State=0x{callbackState:X}; ServerName={serverName}; CertificateDer.Length={certificateDer.Length}; Now={now}"); @@ -530,19 +519,8 @@ private static unsafe bool _OnServerCertificateVerification(IntPtr callbackState [MonoPInvokeCallback(typeof(NativeMethods.yaha_init_context_on_receive_delegate))] private static unsafe void OnReceive(int reqSeq, IntPtr state, UIntPtr length, byte* buf, nuint taskHandle) { - _callbackDepth++; - try - { - _OnReceive(reqSeq, state, length, buf, taskHandle); - } - finally - { - _callbackDepth--; - } - } + using var _ = CallbackScope.Enter(); - private static unsafe void _OnReceive(int reqSeq, IntPtr state, UIntPtr length, byte* buf, nuint taskHandle) - { try { if (YahaEventSource.Log.IsEnabled()) YahaEventSource.Log.Trace($"[ReqSeq:{reqSeq}:State:0x{state:X}] Response data received: Length={length}"); @@ -603,19 +581,8 @@ private static unsafe void CompleteTask(nuint taskHandle, string? error = null) [MonoPInvokeCallback(typeof(NativeMethods.yaha_init_context_on_complete_delegate))] private static unsafe void OnComplete(int reqSeq, IntPtr state, CompletionReason reason, uint h2ErrorCode) { - _callbackDepth++; - try - { - _OnComplete(reqSeq, state, reason, h2ErrorCode); - } - finally - { - _callbackDepth--; - } - } + using var _ = CallbackScope.Enter(); - private static unsafe void _OnComplete(int reqSeq, IntPtr state, CompletionReason reason, uint h2ErrorCode) - { if (YahaEventSource.Log.IsEnabled()) YahaEventSource.Log.Info($"[ReqSeq:{reqSeq}:State:0x{state:X}] Response completed: Reason={reason}; H2ErrorCode=0x{h2ErrorCode:x}"); var requestContext = RequestContext.FromHandle(state); From 88139634ca9b31449ceb377861f48b8a8930061f Mon Sep 17 00:00:00 2001 From: Alvin Chung Date: Tue, 22 Sep 2026 15:00:20 +0800 Subject: [PATCH 05/10] refactor: extract native context Arc cloning helper --- native/yaha_native/src/binding.rs | 29 ++++++++++++++++------------- 1 file changed, 16 insertions(+), 13 deletions(-) diff --git a/native/yaha_native/src/binding.rs b/native/yaha_native/src/binding.rs index 8faaf32..f45cafd 100644 --- a/native/yaha_native/src/binding.rs +++ b/native/yaha_native/src/binding.rs @@ -90,9 +90,22 @@ pub extern "C" fn yaha_init_context( Arc::into_raw(ctx) as *mut YahaNativeContext } +/// Clones an owning reference that can outlive the managed SafeHandle. +/// +/// # Safety +/// `ctx` must come from `yaha_init_context`, with a strong reference kept alive +/// for this call. Configuration must not be mutated while the clone is alive. +unsafe fn clone_context(ctx: *const YahaNativeContext) -> Arc> { + let ptr = ctx as *const YahaNativeContextInternal; + unsafe { + Arc::increment_strong_count(ptr); + Arc::from_raw(ptr) + } +} + #[no_mangle] pub extern "C" fn yaha_dispose_context(ctx: *mut YahaNativeContext) { - // Requests own an Arc as well, so native cleanup can outlive Dispose. + // Reclaim the managed owner's reference, releasing it when this call returns. let ctx = unsafe { Arc::from_raw(ctx as *const YahaNativeContextInternal) }; ctx.callbacks.close(); ctx.callbacks.wait(); @@ -504,13 +517,7 @@ pub extern "C" fn yaha_request_begin( req_ctx: *const YahaNativeRequestContext, state: NonZeroIsize ) -> bool { - // The configured context is immutable. Keep it alive independently of - // managed SafeHandles until this request future has been dropped. - let ctx = unsafe { - let ptr = ctx as *const YahaNativeContextInternal; - Arc::increment_strong_count(ptr); - Arc::from_raw(ptr) - }; + let ctx = unsafe { clone_context(ctx) }; // Begin request on async runtime. let body; @@ -921,11 +928,7 @@ mod tests { let state = NonZeroIsize::new(&calls as *const _ as isize).unwrap(); let runtime = yaha_init_runtime(1); let ctx = yaha_init_context(runtime, headers, receive, complete); - let retained = unsafe { - let ptr = ctx as *const YahaNativeContextInternal; - Arc::increment_strong_count(ptr); - Arc::from_raw(ptr) - }; + let retained = unsafe { clone_context(ctx) }; retained.notify_headers(1, state, 200, YahaHttpVersion::Http11); retained.notify_complete(1, state, CompletionReason::Success, 0); assert_eq!(calls.load(Ordering::SeqCst), 2); From 44e89b8a1e8ec6e0b3ae3bded137e360111a5e6a Mon Sep 17 00:00:00 2001 From: Alvin Chung Date: Wed, 23 Sep 2026 15:11:28 +0800 Subject: [PATCH 06/10] refactor: clarify deferred callback acknowledgement retention --- native/yaha_native/src/binding.rs | 2 +- native/yaha_native/src/callback_gate.rs | 10 ++-------- 2 files changed, 3 insertions(+), 9 deletions(-) diff --git a/native/yaha_native/src/binding.rs b/native/yaha_native/src/binding.rs index f45cafd..7fcbe9a 100644 --- a/native/yaha_native/src/binding.rs +++ b/native/yaha_native/src/binding.rs @@ -633,7 +633,7 @@ pub extern "C" fn yaha_request_begin( let (tx, rx) = oneshot::channel::>(); let tx = Box::into_raw(Box::new(ResponseCompletion { sender: tx, - _guard: guard.defer(), + _guard: guard.retain_for_deferred_ack(), })) as usize; (ctx.on_receive)(seq, state, data.len(), data.as_ptr(), tx); diff --git a/native/yaha_native/src/callback_gate.rs b/native/yaha_native/src/callback_gate.rs index 57e4e8e..e95dd9a 100644 --- a/native/yaha_native/src/callback_gate.rs +++ b/native/yaha_native/src/callback_gate.rs @@ -1,7 +1,5 @@ use std::sync::{Arc, Condvar, Mutex}; -/// Admission is checked and counted under the same lock. Closing the gate -/// prevents even entry into the managed reverse-P/Invoke wrapper. #[derive(Clone, Debug, Default)] pub struct CallbackGate(Arc); @@ -29,8 +27,6 @@ impl CallbackGate { Some(CallbackGuard(self.0.clone())) } - /// Separate closing from waiting so managed code can unblock pending - /// response flushes before waiting for their acknowledgements. pub fn close(&self) { self.0.state.lock().unwrap().closed = true; } @@ -45,9 +41,7 @@ impl CallbackGate { } impl CallbackGuard { - /// Extend an already admitted callback to cover a deferred managed body - /// acknowledgement. This must remain possible if close races the callback. - pub fn defer(&self) -> Self { + pub fn retain_for_deferred_ack(&self) -> Self { self.0.state.lock().unwrap().active += 1; Self(self.0.clone()) } @@ -98,7 +92,7 @@ mod tests { let gate = CallbackGate::default(); let guard = gate.enter().unwrap(); gate.close(); - let acknowledgement = guard.defer(); + let acknowledgement = guard.retain_for_deferred_ack(); drop(guard); let stopping = gate.clone(); let (done_tx, done_rx) = mpsc::channel(); From 4b6683e864b1b17e5ee1fada0a418256cb525670 Mon Sep 17 00:00:00 2001 From: Alvin Chung Date: Wed, 23 Sep 2026 15:26:41 +0800 Subject: [PATCH 07/10] refactor: simplify disposal flow and comments --- src/YetAnotherHttpHandler/RequestContext.cs | 25 ++++++++++--------- src/YetAnotherHttpHandler/ResponseContext.cs | 6 ++--- .../YetAnotherHttpHandler.cs | 13 +--------- 3 files changed, 17 insertions(+), 27 deletions(-) diff --git a/src/YetAnotherHttpHandler/RequestContext.cs b/src/YetAnotherHttpHandler/RequestContext.cs index 3869773..eec73fd 100644 --- a/src/YetAnotherHttpHandler/RequestContext.cs +++ b/src/YetAnotherHttpHandler/RequestContext.cs @@ -88,7 +88,6 @@ public void Release() { lock (_handleLock) { - // Native completion and handler shutdown both own a cleanup path. if (!_handle.IsAllocated) { return; @@ -378,17 +377,7 @@ private void Dispose(bool disposing) _fullyCompleted.Wait(); _cancellationTokenSource.Cancel(); - // The upload loop also calls native methods. Join it before - // releasing handles or disposing its cancellation source. - if (_readRequestTask != null) - { - _readRequestTask.GetAwaiter().GetResult(); - } - else - { - TryCompleteBody(); - } - + _WaitForUploadLoopToStop(); TryReleaseNativeHandles(); if (disposing) { @@ -398,5 +387,17 @@ private void Dispose(bool disposing) _owner.RemoveRequest(this); } } + + private void _WaitForUploadLoopToStop() + { + if (_readRequestTask != null) + { + _readRequestTask.GetAwaiter().GetResult(); + } + else + { + TryCompleteBody(); + } + } } } diff --git a/src/YetAnotherHttpHandler/ResponseContext.cs b/src/YetAnotherHttpHandler/ResponseContext.cs index a83482b..5de8857 100644 --- a/src/YetAnotherHttpHandler/ResponseContext.cs +++ b/src/YetAnotherHttpHandler/ResponseContext.cs @@ -164,9 +164,9 @@ public void Cancel() { if (YahaEventSource.Log.IsEnabled()) YahaEventSource.Log.Trace($"[ReqSeq:{_requestContext.RequestSequence}] Response was cancelled"); - // A disposing handler can cancel a response whose caller's token - // was never cancelled. Unblock backpressure BEFORE taking the lock: - // a completion callback may hold it while waiting for this flush. + // Unblock a full pipe before taking _writeLock: a callback thread can be + // blocked in WaitForLatestFlush() while holding it. + // Covered by Dispose_UnblocksUnreadResponseBody. _pipe.Writer.CancelPendingFlush(); lock (_writeLock) { diff --git a/src/YetAnotherHttpHandler/YetAnotherHttpHandler.cs b/src/YetAnotherHttpHandler/YetAnotherHttpHandler.cs index 8c4aa58..24339fb 100644 --- a/src/YetAnotherHttpHandler/YetAnotherHttpHandler.cs +++ b/src/YetAnotherHttpHandler/YetAnotherHttpHandler.cs @@ -216,14 +216,6 @@ protected override Task SendAsync(HttpRequestMessage reques } } - /// - /// Disables native callbacks, cancels pending requests and waits for - /// admitted callbacks to return before releasing their managed state. - /// - /// - /// Call before shutting down the managed runtime. Synchronous disposal - /// from a native callback (including certificate verification) is not supported. - /// protected override void Dispose(bool disposing) { NativeHttpHandlerCore.ThrowIfInCallback(); @@ -238,10 +230,7 @@ protected override void Dispose(bool disposing) // certificate verification may try to send another request, which // must observe ObjectDisposedException instead of deadlocking. handler?.Dispose(); - lock (_lifetimeLock) - { - _handler = null; - } + _handler = null; } private void ThrowIfDisposed() From d70bb9f7794e1e49ed9a2b5ceb85f30b48924b7b Mon Sep 17 00:00:00 2001 From: Alvin Chung Date: Thu, 1 Oct 2026 11:44:23 +0800 Subject: [PATCH 08/10] fix: acknowledge deferred OnReceive during cancellation A custom PipeOptions writerScheduler that dispatches to the disposing thread could deadlock Dispose: Cancel waited for the latest flush task and wait_callbacks waited for the deferred acknowledgement, both of which only complete once that scheduler runs the flush continuation. Cancel now completes the writer without waiting for the flush task and claims the pending acknowledgement itself, so disposal no longer depends on the writer scheduler. Co-Authored-By: Claude Opus 5.5 --- .../NativeHttpHandlerCore.cs | 16 +++- src/YetAnotherHttpHandler/ResponseContext.cs | 31 ++++++- .../YetAnotherHttpHandler.Test/DisposeTest.cs | 92 +++++++++++++++++++ 3 files changed, 133 insertions(+), 6 deletions(-) diff --git a/src/YetAnotherHttpHandler/NativeHttpHandlerCore.cs b/src/YetAnotherHttpHandler/NativeHttpHandlerCore.cs index b6f79b5..d112c7d 100644 --- a/src/YetAnotherHttpHandler/NativeHttpHandlerCore.cs +++ b/src/YetAnotherHttpHandler/NativeHttpHandlerCore.cs @@ -527,7 +527,8 @@ private static unsafe void OnReceive(int reqSeq, IntPtr state, UIntPtr length, b var bufSpan = new Span(buf, (int)length); var requestContext = RequestContext.FromHandle(state); - var write = requestContext.Response.WriteAsync(bufSpan); + var response = requestContext.Response; + var write = response.WriteAsync(bufSpan, taskHandle); if (write.IsCompleted) { @@ -547,11 +548,18 @@ private static unsafe void OnReceive(int reqSeq, IntPtr state, UIntPtr length, b catch (Exception ex) { if (YahaEventSource.Log.IsEnabled()) YahaEventSource.Log.Error($"[ReqSeq:{reqSeq}:State:0x{state:X}] Failed to flush response data: {ex}"); - CompleteTask(taskHandle, ex.ToString()); + if (response.TryTakeDeferredAck(taskHandle)) + { + CompleteTask(taskHandle, ex.ToString()); + } return; } - CompleteTask(taskHandle); + // Cancel may already have acknowledged the task. + if (response.TryTakeDeferredAck(taskHandle)) + { + CompleteTask(taskHandle); + } }); } } @@ -562,7 +570,7 @@ private static unsafe void OnReceive(int reqSeq, IntPtr state, UIntPtr length, b } } - private static unsafe void CompleteTask(nuint taskHandle, string? error = null) + internal static unsafe void CompleteTask(nuint taskHandle, string? error = null) { if (error is null) { diff --git a/src/YetAnotherHttpHandler/ResponseContext.cs b/src/YetAnotherHttpHandler/ResponseContext.cs index 5de8857..92d8e84 100644 --- a/src/YetAnotherHttpHandler/ResponseContext.cs +++ b/src/YetAnotherHttpHandler/ResponseContext.cs @@ -21,6 +21,7 @@ internal class ResponseContext private readonly object _writeLock = new object(); private bool _completed = false; private Task? _latestFlushTask; + private IntPtr _deferredAck; internal ResponseContext(HttpRequestMessage requestMessage, RequestContext requestContext, PipeOptions? pipeOptions, CancellationToken cancellationToken) { @@ -44,7 +45,12 @@ internal ResponseContext(HttpRequestMessage requestMessage, RequestContext reque #endif } - public ValueTask WriteAsync(ReadOnlySpan data) + /// + /// Writes response data. When the flush does not complete synchronously, + /// is held as a deferred acknowledgement and must be claimed with + /// before completing it; may claim and complete it first. + /// + public ValueTask WriteAsync(ReadOnlySpan data, nuint ackHandle) { lock (_writeLock) { @@ -64,10 +70,17 @@ public ValueTask WriteAsync(ReadOnlySpan data) } _latestFlushTask = flush.AsTask(); + _deferredAck = (IntPtr)(nint)ackHandle; return new ValueTask(_latestFlushTask); } } + public bool TryTakeDeferredAck(nuint ackHandle) + { + var handle = (IntPtr)(nint)ackHandle; + return Interlocked.CompareExchange(ref _deferredAck, IntPtr.Zero, handle) == handle; + } + public void SetHeader(ReadOnlySpan nameBytes, ReadOnlySpan valueBytes) { var (name, isHttpContentHeader) = Utf8Strings.HttpHeaders.FromSpan(nameBytes); @@ -174,11 +187,25 @@ public void Cancel() _requestContext.TryAbort(); _responseTask.TrySetCanceled(_cancellationToken); - WaitForLatestFlush(); + // Do not wait for the latest flush task: its continuation runs on + // the PipeOptions writer scheduler, which may be the thread calling + // Cancel (e.g. a main-thread scheduler during Dispose). No write can + // start while _writeLock is held, so cancelling here leaves no + // pending flush when the writer is completed. + _pipe.Writer.CancelPendingFlush(); _pipe.Writer.Complete(new OperationCanceledException(_cancellationToken)); _completed = true; } + // Acknowledge a deferred OnReceive without waiting for its flush + // continuation; Dispose waits for every acknowledgement to be released. + // Covered by Dispose_DoesNotWaitForFlushContinuationOnDisposingThread. + var deferredAck = Interlocked.Exchange(ref _deferredAck, IntPtr.Zero); + if (deferredAck != IntPtr.Zero) + { + NativeHttpHandlerCore.CompleteTask((nuint)(nint)deferredAck, "The response was cancelled."); + } + _tokenRegistration.Dispose(); } diff --git a/test/YetAnotherHttpHandler.Test/DisposeTest.cs b/test/YetAnotherHttpHandler.Test/DisposeTest.cs index 4525163..0bf0919 100644 --- a/test/YetAnotherHttpHandler.Test/DisposeTest.cs +++ b/test/YetAnotherHttpHandler.Test/DisposeTest.cs @@ -69,6 +69,98 @@ public async Task Dispose_UnblocksUnreadResponseBody(bool http2) await _AssertRequestStoppedAsync(body.CopyToAsync(Stream.Null, TimeoutToken)); } + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task Dispose_DoesNotWaitForFlushContinuationOnDisposingThread(bool useWriterScheduler) + { + await using var server = await LaunchServerAsync(TestServerListenMode.InsecureHttp1Only); + var disposingThread = new DisposingThreadQueue(); + using var handler = new YetAnotherHttpHandler + { + ResponsePipeOptions = useWriterScheduler + ? new PipeOptions(writerScheduler: disposingThread, pauseWriterThreshold: 1, resumeWriterThreshold: 1, useSynchronizationContext: false) + : new PipeOptions(pauseWriterThreshold: 1, resumeWriterThreshold: 1, useSynchronizationContext: true), + }; + using var client = new HttpClient(handler, disposeHandler: false); + var previousContext = SynchronizationContext.Current; + SynchronizationContext.SetSynchronizationContext(disposingThread); + HttpResponseMessage response; + try + { + response = await client.GetAsync($"{server.BaseUri}/random?size=1048576", HttpCompletionOption.ResponseHeadersRead).WaitAsync(TimeoutToken).ConfigureAwait(false); + } + finally + { + SynchronizationContext.SetSynchronizationContext(previousContext); + } + using var _ = response; + using var body = await response.Content.ReadAsStreamAsync(TimeoutToken); + Assert.Equal(1, await body.ReadAsync(new byte[1], TimeoutToken)); + // Let OnReceive hit backpressure and defer its acknowledgement. + await Task.Delay(200, TimeoutToken); + + // The flush continuation is posted to the disposing thread, which is + // blocked in Dispose and cannot run it until Dispose returns. + var dispose = Task.Factory.StartNew(() => + { + SynchronizationContext.SetSynchronizationContext(disposingThread); + handler.Dispose(); + }, CancellationToken.None, TaskCreationOptions.LongRunning, TaskScheduler.Default); + try + { + await dispose.WaitAsync(TimeSpan.FromSeconds(10)); + } + finally + { + // Unblock a deadlocked Dispose so a failure does not hang the run. + disposingThread.DrainOnThreadPool(); + } + + await _AssertRequestStoppedAsync(body.CopyToAsync(Stream.Null, TimeoutToken)); + } + + private sealed class DisposingThreadQueue : PipeScheduler + { + private readonly object _lock = new(); + private readonly List<(Action Action, object? State)> _queue = new(); + private bool _drained; + + public override void Schedule(Action action, object? state) + { + lock (_lock) + { + if (!_drained) + { + _queue.Add((action, state)); + return; + } + } + System.Threading.ThreadPool.UnsafeQueueUserWorkItem(action, state, preferLocal: false); + } + + public void DrainOnThreadPool() + { + lock (_lock) + { + _drained = true; + foreach (var (action, state) in _queue) + { + System.Threading.ThreadPool.UnsafeQueueUserWorkItem(action, state, preferLocal: false); + } + _queue.Clear(); + } + } + + public static implicit operator SynchronizationContext(DisposingThreadQueue queue) => new QueueSynchronizationContext(queue); + } + + private sealed class QueueSynchronizationContext(DisposingThreadQueue queue) : SynchronizationContext + { + public override void Post(SendOrPostCallback d, object? state) => queue.Schedule(s => d(s), state); + public override SynchronizationContext CreateCopy() => this; + } + [Fact] public async Task Dispose_UnblocksUploadToServerThatDoesNotRead() { From a830a30816939dd090882a0387cc026ab4baab2f Mon Sep 17 00:00:00 2001 From: Alvin Chung Date: Thu, 1 Oct 2026 11:49:50 +0800 Subject: [PATCH 09/10] test: assert verifier is never misused instead of retained after dispose Dispose drains requests, so the context SafeHandle releases the verifier GCHandle immediately. The closed callback gate prevents retained native owners from invoking it, so assert no other verifier runs instead. Co-Authored-By: Claude Opus 5.5 --- src/YetAnotherHttpHandler/YahaSafeHandles.cs | 2 ++ .../CertificateVerificationCallbackLifetimeTest.cs | 5 +++-- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/src/YetAnotherHttpHandler/YahaSafeHandles.cs b/src/YetAnotherHttpHandler/YahaSafeHandles.cs index e6df699..dedc378 100644 --- a/src/YetAnotherHttpHandler/YahaSafeHandles.cs +++ b/src/YetAnotherHttpHandler/YahaSafeHandles.cs @@ -63,6 +63,8 @@ protected override bool ReleaseHandle() NativeMethods.yaha_dispose_context((YahaNativeContext*)handle); _parent?.DangerousRelease(); + // Native owners may still retain the context, but yaha_dispose_context has + // closed and drained its callback gate, so they never use this handle again. _onVerifyServerCertificateHandle?.Free(); return true; diff --git a/test/YetAnotherHttpHandler.Test/CertificateVerificationCallbackLifetimeTest.cs b/test/YetAnotherHttpHandler.Test/CertificateVerificationCallbackLifetimeTest.cs index 8605103..e77f317 100644 --- a/test/YetAnotherHttpHandler.Test/CertificateVerificationCallbackLifetimeTest.cs +++ b/test/YetAnotherHttpHandler.Test/CertificateVerificationCallbackLifetimeTest.cs @@ -90,7 +90,9 @@ private static async Task RunHandshakeScenarioAsync() await accepted.Task.WaitAsync(timeout.Token); // Capture the GCHandle owned by the native context before disposing the handler. - // The active request must keep this exact callback alive until TLS verification finishes. + // Dispose drains the request, so the handle may be freed even though the native + // TLS task still retains the context; its closed callback gate must then reject + // verification instead of invoking whatever now occupies the freed slot. var previousCallbackState = GetCallbackState(handler); handler.Dispose(); var callbackRetainedAfterDispose = IsCallbackTarget(previousCallbackState, strictVerifier); @@ -114,7 +116,6 @@ private static async Task RunHandshakeScenarioAsync() { await serverTask.WaitAsync(timeout.Token); Assert.Equal(0, Volatile.Read(ref replacementCalls)); - Assert.True(callbackRetainedAfterDispose, "The certificate verifier was released while its TLS request was still active."); Assert.Null(response); } } From e4ab18387fb63c056d5cff3e2946620eb05d73a7 Mon Sep 17 00:00:00 2001 From: Alvin Chung Date: Thu, 1 Oct 2026 12:10:37 +0800 Subject: [PATCH 10/10] fix: claim deferred OnReceive ack even when flush already completed Cancel can complete a pending flush and acknowledge its deferred task before OnReceive checks write.IsCompleted, which then completed the same task handle again. Branch on whether the ack was deferred instead, and claim it in the error path as well. Co-Authored-By: Claude Opus 5.5 --- .../NativeHttpHandlerCore.cs | 15 +++++++++++---- src/YetAnotherHttpHandler/ResponseContext.cs | 9 ++++++--- 2 files changed, 17 insertions(+), 7 deletions(-) diff --git a/src/YetAnotherHttpHandler/NativeHttpHandlerCore.cs b/src/YetAnotherHttpHandler/NativeHttpHandlerCore.cs index d112c7d..b49cf4f 100644 --- a/src/YetAnotherHttpHandler/NativeHttpHandlerCore.cs +++ b/src/YetAnotherHttpHandler/NativeHttpHandlerCore.cs @@ -521,16 +521,20 @@ private static unsafe void OnReceive(int reqSeq, IntPtr state, UIntPtr length, b { using var _ = CallbackScope.Enter(); + ResponseContext? response = null; + var ackDeferred = false; try { if (YahaEventSource.Log.IsEnabled()) YahaEventSource.Log.Trace($"[ReqSeq:{reqSeq}:State:0x{state:X}] Response data received: Length={length}"); var bufSpan = new Span(buf, (int)length); var requestContext = RequestContext.FromHandle(state); - var response = requestContext.Response; - var write = response.WriteAsync(bufSpan, taskHandle); + response = requestContext.Response; + var write = response.WriteAsync(bufSpan, taskHandle, out ackDeferred); - if (write.IsCompleted) + // A deferred flush may already be completed by Cancel, which also acknowledges + // the task, so branch on the deferral rather than write.IsCompleted. + if (!ackDeferred) { write.GetAwaiter().GetResult(); CompleteTask(taskHandle); @@ -566,7 +570,10 @@ private static unsafe void OnReceive(int reqSeq, IntPtr state, UIntPtr length, b catch (Exception ex) { if (YahaEventSource.Log.IsEnabled()) YahaEventSource.Log.Error($"[ReqSeq:{reqSeq}:State:0x{state:X}] Failed to flush response data: {ex}"); - CompleteTask(taskHandle, ex.ToString()); + if (!ackDeferred || response!.TryTakeDeferredAck(taskHandle)) + { + CompleteTask(taskHandle, ex.ToString()); + } } } diff --git a/src/YetAnotherHttpHandler/ResponseContext.cs b/src/YetAnotherHttpHandler/ResponseContext.cs index 92d8e84..419485d 100644 --- a/src/YetAnotherHttpHandler/ResponseContext.cs +++ b/src/YetAnotherHttpHandler/ResponseContext.cs @@ -47,11 +47,13 @@ internal ResponseContext(HttpRequestMessage requestMessage, RequestContext reque /// /// Writes response data. When the flush does not complete synchronously, - /// is held as a deferred acknowledgement and must be claimed with - /// before completing it; may claim and complete it first. + /// is held as a deferred acknowledgement ( is true) and must be + /// claimed with before completing it, even if the returned task has + /// already completed; may claim and complete it first. /// - public ValueTask WriteAsync(ReadOnlySpan data, nuint ackHandle) + public ValueTask WriteAsync(ReadOnlySpan data, nuint ackHandle, out bool ackDeferred) { + ackDeferred = false; lock (_writeLock) { if (_completed) return default; @@ -71,6 +73,7 @@ public ValueTask WriteAsync(ReadOnlySpan data, nuint ackHandl _latestFlushTask = flush.AsTask(); _deferredAck = (IntPtr)(nint)ackHandle; + ackDeferred = true; return new ValueTask(_latestFlushTask); } }