From b4085284ad9124af3665a92b9f3c38eeffd86627 Mon Sep 17 00:00:00 2001 From: Ruslan Pislari Date: Wed, 17 Jun 2026 10:33:33 +0300 Subject: [PATCH 1/4] feat: add YamuxUnixSocket implementation for independent evolution of yamux transport --- Cargo.lock | 182 +----- nginx_module/Cargo.toml | 14 +- nginx_module/src/lib.rs | 3 + nginx_module/src/yamux_unix_socket.rs | 767 ++++++++++++++++++++++++++ 4 files changed, 788 insertions(+), 178 deletions(-) create mode 100644 nginx_module/src/yamux_unix_socket.rs diff --git a/Cargo.lock b/Cargo.lock index bd3be25..ca2d68e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -19,16 +19,14 @@ checksum = "a23eb6b1614318a8071c9b2521f36b424b2c83db5eb3a0fead4a6c0809af6e61" [[package]] name = "bindgen" -version = "0.69.5" +version = "0.72.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "271383c67ccabffb7381723dea0672a673f292304fcb45c01cc648c7a8d58088" +checksum = "993776b509cfb49c750f11b8f07a46fa23e0a1386ffc01fb1e7d343efc387895" dependencies = [ "bitflags", "cexpr", "clang-sys", "itertools", - "lazy_static", - "lazycell", "log", "prettyplease", "proc-macro2", @@ -37,14 +35,13 @@ dependencies = [ "rustc-hash", "shlex", "syn", - "which", ] [[package]] name = "bitflags" -version = "2.10.0" +version = "2.13.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "812e12b5285cc515a9c72a5c1d3b6d46a19dac5acfef5265968c166106e31dd3" +checksum = "b4388bee8683e3d04af747c73422af53102d2bd24d9eadb6cbc100baef4b43f8" [[package]] name = "cexpr" @@ -89,16 +86,6 @@ version = "1.15.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "48c757948c5ede0e46177b7add2e67155f70e33c07fea8284df6576da70b3719" -[[package]] -name = "errno" -version = "0.3.14" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" -dependencies = [ - "libc", - "windows-sys 0.61.2", -] - [[package]] name = "form_urlencoded" version = "1.2.2" @@ -120,15 +107,6 @@ version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" -[[package]] -name = "home" -version = "0.5.11" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "589533453244b0995c858700322199b2becb13b627df2851f64a2775d024abcf" -dependencies = [ - "windows-sys 0.59.0", -] - [[package]] name = "icu_collections" version = "2.0.0" @@ -245,18 +223,6 @@ dependencies = [ "either", ] -[[package]] -name = "lazy_static" -version = "1.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" - -[[package]] -name = "lazycell" -version = "1.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "830d08ce1d1d941e6b30645f1a0eb5643013d835ce3779a5fc208261dbe10f55" - [[package]] name = "libc" version = "0.2.177" @@ -273,12 +239,6 @@ dependencies = [ "windows-link", ] -[[package]] -name = "linux-raw-sys" -version = "0.4.15" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d26c52dbd32dccf2d10cac7725f8eae5296885fb5703b261f7d0a0739ec807ab" - [[package]] name = "litemap" version = "0.8.0" @@ -314,7 +274,7 @@ dependencies = [ [[package]] name = "nginx_module" -version = "0.2.3" +version = "0.3.0" dependencies = [ "anyhow", "bindgen", @@ -334,12 +294,6 @@ dependencies = [ "minimal-lexical", ] -[[package]] -name = "once_cell" -version = "1.21.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "42f5e15c9953c5e4ccceeb2e7382a716482c34515315f7b03532b8b4e8393d2d" - [[package]] name = "percent-encoding" version = "2.3.2" @@ -414,28 +368,9 @@ checksum = "7a2d987857b319362043e95f5353c0535c1f58eec5336fdfcf626430af7def58" [[package]] name = "rustc-hash" -version = "1.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "08d43f7aa6b08d49f382cde6a7982047c3426db949b1424bc4b7ec9ae12c6ce2" - -[[package]] -name = "rustix" -version = "0.38.44" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fdb5bc1ae2baa591800df16c9ca78619bf65c0488b41b96ccec5d11220d8c154" -dependencies = [ - "bitflags", - "errno", - "libc", - "linux-raw-sys", - "windows-sys 0.59.0", -] - -[[package]] -name = "rustversion" -version = "1.0.22" +version = "2.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d" +checksum = "94300abf3f1ae2e2b8ffb7b58043de3d399c73fa6f4b73826402a5c457614dbe" [[package]] name = "serde" @@ -496,23 +431,22 @@ checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" [[package]] name = "strum" -version = "0.26.3" +version = "0.28.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8fec0f0aef304996cf250b31b5a10dee7980c85da9d759361292b8bca5a18f06" +checksum = "9628de9b8791db39ceda2b119bbe13134770b56c138ec1d3af810d045c04f9bd" dependencies = [ "strum_macros", ] [[package]] name = "strum_macros" -version = "0.26.4" +version = "0.28.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4c6bee85a5a24955dc440386795aa378cd9cf82acd5f764469152d2270e581be" +checksum = "ab85eea0270ee17587ed4156089e10b9e6880ee688791d45a905f5b1ca36f664" dependencies = [ "heck", "proc-macro2", "quote", - "rustversion", "syn", ] @@ -572,106 +506,12 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be" -[[package]] -name = "which" -version = "4.4.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "87ba24419a2078cd2b0f2ede2691b6c66d8e47836da3b6db8265ebad47afbfc7" -dependencies = [ - "either", - "home", - "once_cell", - "rustix", -] - [[package]] name = "windows-link" version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" -[[package]] -name = "windows-sys" -version = "0.59.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1e38bc4d79ed67fd075bcc251a1c39b32a1776bbe92e5bef1f0bf1f8c531853b" -dependencies = [ - "windows-targets", -] - -[[package]] -name = "windows-sys" -version = "0.61.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" -dependencies = [ - "windows-link", -] - -[[package]] -name = "windows-targets" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" -dependencies = [ - "windows_aarch64_gnullvm", - "windows_aarch64_msvc", - "windows_i686_gnu", - "windows_i686_gnullvm", - "windows_i686_msvc", - "windows_x86_64_gnu", - "windows_x86_64_gnullvm", - "windows_x86_64_msvc", -] - -[[package]] -name = "windows_aarch64_gnullvm" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" - -[[package]] -name = "windows_aarch64_msvc" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" - -[[package]] -name = "windows_i686_gnu" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b" - -[[package]] -name = "windows_i686_gnullvm" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" - -[[package]] -name = "windows_i686_msvc" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" - -[[package]] -name = "windows_x86_64_gnu" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" - -[[package]] -name = "windows_x86_64_gnullvm" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" - -[[package]] -name = "windows_x86_64_msvc" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" - [[package]] name = "writeable" version = "0.6.1" diff --git a/nginx_module/Cargo.toml b/nginx_module/Cargo.toml index 18e83c1..3e18a95 100644 --- a/nginx_module/Cargo.toml +++ b/nginx_module/Cargo.toml @@ -8,17 +8,17 @@ name = "nginx_module" readme = "README.md" repository = "https://github.com/g-Core/nginx-rust" - version = "0.2.3" + version = "0.3.0" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [build-dependencies] -bindgen = "0.69.4" +bindgen = "0.72" [dependencies] -anyhow = "1.0.71" -bitflags = "2.3.1" -libc = "0.2.144" -strum = { version = "0.26.1", features = ["derive"] } -url = "2.5.0" \ No newline at end of file +anyhow = "1.0" +bitflags = "2.13" +libc = "0.2" +strum = { version = "0.28", features = ["derive"] } +url = "2.5" \ No newline at end of file diff --git a/nginx_module/src/lib.rs b/nginx_module/src/lib.rs index 4b1592d..b68bb7c 100644 --- a/nginx_module/src/lib.rs +++ b/nginx_module/src/lib.rs @@ -60,6 +60,9 @@ pub use wrappers::{hex_dump, IndexedVar, NgxConfig}; mod unix_socket; pub use unix_socket::{Disconnected, UnixSocket}; +mod yamux_unix_socket; +pub use yamux_unix_socket::YamuxUnixSocket; + mod timer; pub use timer::Timer; diff --git a/nginx_module/src/yamux_unix_socket.rs b/nginx_module/src/yamux_unix_socket.rs new file mode 100644 index 0000000..8103029 --- /dev/null +++ b/nginx_module/src/yamux_unix_socket.rs @@ -0,0 +1,767 @@ +//! A standalone duplicate of [`crate::unix_socket`] for the yamux daemon proxy. +//! +//! This intentionally duplicates the `UnixSocket` implementation so the yamux +//! transport can evolve independently of the one used by the v2a/v2b daemon +//! proxies. It is currently a faithful copy exposing the type as +//! [`YamuxUnixSocket`]: it performs a connection-level handshake on connect, then +//! hands every subsequent inbound byte to `read_event` (which feeds the yamux +//! session). `after_handshake` fires on each successful (re)connect so the caller +//! can (re)initialise per-connection state — e.g. rebuild the yamux session. + +use libc::{c_void, sockaddr_un}; +use std::{ + borrow::Cow, cell::RefCell, collections::VecDeque, marker::PhantomPinned, mem::MaybeUninit, + os::unix::ffi::OsStrExt, path::Path, pin::Pin, +}; + +use crate::log::ngx_log_error; +use crate::{ + bindings::{ + ngx_close_connection, ngx_connection_t, ngx_cycle, ngx_event_actions, + ngx_event_connect_peer, ngx_event_t, ngx_exiting, ngx_handle_read_event, + ngx_handle_write_event, ngx_log_t, ngx_peer_connection_t, ngx_quit, ngx_terminate, + NGX_AGAIN, NGX_LOG_ERR, NGX_RS_WRITE_EVENT, + }, + ngx_event_add_timer, ngx_event_del_timer, NgxStr, NGX_OK, +}; + +pub struct YamuxUnixSocket(Pin>); + +enum State { + Connected { + conn: *mut ngx_connection_t, + buffers: WriteBuffers, + }, + WaitServerHandshake { + conn: *mut ngx_connection_t, + buffers: WriteBuffers, + }, + Disconnected { + event: Box, + reconnect_timeout: usize, + }, +} + +struct Inner { + state: RefCell, + on_read: Box>, + check_handshake: Box>, + after_handshake: Box Vec + 'static>>, + handshake_msg: Vec, + path: String, + name: Cow<'static, [u8]>, + dummy_log: Box, + _phantom: PhantomPinned, +} + +type ReadFn = dyn FnMut(&[u8]) -> Vec; +type ValidateFn = dyn FnMut(&[u8]) -> anyhow::Result<()>; + +#[derive(Default)] +struct WriteBuffers { + buffers: VecDeque, +} + +struct WriteBuffer { + data: Box<[u8]>, + start: u32, + end: u32, +} + +const MIN_TIMEOUT_MS: usize = 1; +const TIMEOUT_MS: usize = 255; + +fn log_borrow_mut_error(context: &str, error: std::cell::BorrowMutError) { + if let Ok(err_msg) = std::ffi::CString::new(format!("{}: {}", context, error)) { + unsafe { ngx_log_error(NGX_LOG_ERR as usize, (*ngx_cycle).log, 0, &err_msg) }; + } +} + +impl YamuxUnixSocket { + pub fn connect( + path: String, + name: Cow<'static, [u8]>, + read_event: impl FnMut(&[u8]) -> Vec + 'static, + handshake_msg: Vec, + check_server_handshake: impl FnMut(&[u8]) -> anyhow::Result<()> + 'static, + after_handshake: impl FnMut() -> Vec + 'static, + ) -> Self { + let mut dummy_log: Box = + Box::new(unsafe { MaybeUninit::zeroed().assume_init() }); + dummy_log.writer = Some(dummy_log_fn); + let state = match State::try_connect(&path, &name, &mut *dummy_log) { + Some(conn) => { + let mut buffers = WriteBuffers::default(); + buffers.push(&handshake_msg); + match unsafe { buffers.send(conn) } { + Ok(()) => State::WaitServerHandshake { conn, buffers }, + Err(_) => unsafe { + let mut ev: Box = + Box::new(MaybeUninit::zeroed().assume_init()); + ev.handler = Some(on_reconnect_timeout); + ev.log = (*ngx_cycle).log; + if (*ev).timer_set() == 0 + && ngx_quit == 0 + && ngx_exiting == 0 + && ngx_terminate == 0 + { + ngx_event_add_timer(&mut *ev, MIN_TIMEOUT_MS); + } + State::Disconnected { + event: ev, + reconnect_timeout: MIN_TIMEOUT_MS, + } + }, + } + } + None => unsafe { + let mut ev: Box = Box::new(MaybeUninit::zeroed().assume_init()); + ev.handler = Some(on_reconnect_timeout); + ev.log = (*ngx_cycle).log; + if (*ev).timer_set() == 0 && ngx_quit == 0 && ngx_exiting == 0 && ngx_terminate == 0 + { + ngx_event_add_timer(&mut *ev, MIN_TIMEOUT_MS); + } + State::Disconnected { + event: ev, + reconnect_timeout: MIN_TIMEOUT_MS, + } + }, + }; + + let inner = Inner { + state: RefCell::new(state), + on_read: Box::new(RefCell::new(read_event)), + check_handshake: Box::new(RefCell::new(check_server_handshake)), + after_handshake: Box::new(RefCell::new(after_handshake)), + handshake_msg, + path, + name, + dummy_log, + _phantom: PhantomPinned, + }; + + let inner = Box::pin(inner); + match inner.state.try_borrow_mut() { + Ok(mut state) => match &mut *state { + State::WaitServerHandshake { conn, .. } => unsafe { + (**conn).data = (&*inner as *const Inner as *mut Inner).cast(); + (*(**conn).write).handler = Some(on_write); + (*(**conn).read).handler = Some(on_read); + }, + State::Connected { conn, .. } => unsafe { + (**conn).data = (&*inner as *const Inner as *mut Inner).cast(); + (*(**conn).write).handler = Some(on_write); + (*(**conn).read).handler = Some(on_read); + }, + State::Disconnected { event, .. } => { + event.data = (&*inner as *const Inner as *mut Inner).cast(); + match inner.after_handshake.try_borrow_mut() { + Ok(mut after_handshake) => { + (after_handshake)(); // discard send data, it is disconnected + } + Err(error) => { + log_borrow_mut_error("connect: after_handshake borrow", error); + } + } + } + }, + Err(error) => { + log_borrow_mut_error("connect: state borrow", error); + } + } + + Self(inner) + } + + pub fn disconnected(&self) -> bool { + match self.0.state.try_borrow_mut() { + Ok(state) => matches!(&*state, State::Disconnected { .. }), + Err(error) => { + log_borrow_mut_error("disconnected: state borrow", error); + false + } + } + } + + pub fn stop(&self) { + unsafe { + // create a dummy event, but don't schedule it + let mut dummy_ev: Box = Box::new(MaybeUninit::zeroed().assume_init()); + dummy_ev.handler = None; + dummy_ev.log = (*ngx_cycle).log; + match self.0.state.try_borrow_mut() { + Ok(mut state) => { + *state = State::Disconnected { + event: dummy_ev, + reconnect_timeout: MIN_TIMEOUT_MS, + }; + } + Err(error) => { + log_borrow_mut_error("stop: state borrow", error); + } + } + } + } +} + +impl State { + fn try_connect(path: &str, name: &[u8], log: *mut ngx_log_t) -> Option<*mut ngx_connection_t> { + let mut sockaddr = sockaddr_un { + sun_family: libc::AF_UNIX as libc::sa_family_t, + sun_path: [0; 108], + }; + + let path = Path::new(path); + + let mut len = path.as_os_str().len(); + if len > sockaddr.sun_path.len() { + len = sockaddr.sun_path.len(); + // TODO: report warning + } + + unsafe { + std::ptr::copy_nonoverlapping( + path.as_os_str().as_bytes().as_ptr(), + sockaddr.sun_path.as_mut_ptr().cast(), + len, + ) + }; + + let mut name = NgxStr::from(name); + + let mut pc = ngx_peer_connection_t { + connection: std::ptr::null_mut(), + sockaddr: ((&mut sockaddr) as *mut sockaddr_un).cast(), + socklen: std::mem::size_of_val(&sockaddr) as u32, + name: name.as_mut_ptr(), + tries: 0, + start_time: 0, + get: Some(empty_get), + free: Some(on_free), + notify: None, + data: std::ptr::null_mut(), + set_session: None, + save_session: None, + local: std::ptr::null_mut(), + type_: 0, + rcvbuf: 0, + log, + hint: std::ptr::null_mut(), + sid: std::ptr::null_mut(), + _bitfield_align_1: Default::default(), + _bitfield_1: Default::default(), + __bindgen_padding_0: Default::default(), + }; + + let result = unsafe { ngx_event_connect_peer(&mut pc) }; + + if result != NGX_OK as isize && result != NGX_AGAIN as isize { + None + } else { + let conn = pc.connection; + if conn.is_null() { + None + } else { + Some(conn) + } + } + } +} + +impl Drop for State { + fn drop(&mut self) { + match self { + Self::WaitServerHandshake { conn, .. } => unsafe { + if !(*conn).is_null() && (**conn).destroyed() == 0 { + (**conn).data = std::ptr::null_mut(); + ngx_close_connection(*conn); + } + }, + Self::Connected { conn, .. } => unsafe { + if !(*conn).is_null() && (**conn).destroyed() == 0 { + (**conn).data = std::ptr::null_mut(); + ngx_close_connection(*conn); + } + }, + Self::Disconnected { event, .. } => unsafe { + if event.timer_set() != 0 { + ngx_event_del_timer(event.as_mut()); + } + }, + } + } +} + +unsafe fn conn_data(conn: *mut ngx_connection_t) -> *const Inner { + (*conn).data.cast() +} + +pub struct Disconnected; + +impl YamuxUnixSocket { + pub fn write(&self, buf: &[u8]) -> anyhow::Result<()> { + self.0.write(buf) + } +} + +impl Inner { + fn write(&self, buf: &[u8]) -> anyhow::Result<()> { + match self.state.try_borrow_mut() { + Ok(mut state) => match &mut *state { + State::Connected { + conn, + ref mut buffers, + } => { + buffers.push(buf); + unsafe { + buffers + .send(*conn) + .map_err(|_| anyhow::anyhow!("Disconnected")) + } + } + State::WaitServerHandshake { + conn, + ref mut buffers, + } => { + buffers.push(buf); + unsafe { + buffers + .send(*conn) + .map_err(|_| anyhow::anyhow!("Disconnected")) + } + } + State::Disconnected { .. } => Err(anyhow::anyhow!("Disconnected")), + }, + Err(error) => { + log_borrow_mut_error("write: state borrow", error); + Err(anyhow::anyhow!("Borrow error")) + } + } + } + + // Unsafe - self should come from a Pinned address + unsafe fn send(&self) { + // Use try_borrow_mut to avoid panic on re-entrant calls + let mut state = match self.state.try_borrow_mut() { + Ok(state) => state, + Err(error) => { + if let Ok(err_msg) = std::ffi::CString::new(format!("send: {}", error)) { + ngx_log_error(NGX_LOG_ERR as usize, (*ngx_cycle).log, 0, &err_msg); + } + return; + } // Already borrowed, skip this send attempt + }; + + if let State::Connected { conn, buffers } = &mut *state { + if let Err(Disconnected) = buffers.send(*conn) { + *state = State::Disconnected { + event: self.create_and_schedule_reconnect(), + reconnect_timeout: MIN_TIMEOUT_MS, + }; + } + } + } + + // Unsafe - self should come from a Pined address + unsafe fn create_and_schedule_reconnect(&self) -> Box { + let mut ev: Box = Box::new(MaybeUninit::zeroed().assume_init()); + ev.handler = Some(on_reconnect_timeout); + ev.log = (*ngx_cycle).log; + ev.data = (self as *const Self as *mut Self).cast(); + + if (*ev).timer_set() == 0 && ngx_quit == 0 && ngx_exiting == 0 && ngx_terminate == 0 { + ngx_event_add_timer(&mut *ev, MIN_TIMEOUT_MS); + } + + ev + } +} + +impl WriteBuffers { + fn push(&mut self, data: &[u8]) { + let mut data = data; + if let Some(buf) = self.buffers.back_mut() { + data = buf.push(data); + } + while !data.is_empty() { + let mut buf = WriteBuffer::new(); + data = buf.push(data); + self.buffers.push_back(buf); + } + } + + unsafe fn send(&mut self, conn: *mut ngx_connection_t) -> Result<(), Disconnected> { + if let Some(send) = (*conn).send { + while let Some(first) = self.buffers.front_mut() { + if first.start == first.end { + break; // buffer empty + } + match first.send(conn, send) { + BufferSendResult::Ok => {} + BufferSendResult::EndOfStream => { + return Err(Disconnected); + } + BufferSendResult::Again => { + if (*(*conn).write).timer_set() == 0 + && ngx_quit == 0 + && ngx_exiting == 0 + && ngx_terminate == 0 + { + ngx_event_add_timer((*conn).write, TIMEOUT_MS); + } + if ngx_handle_write_event((*conn).write, 0) != NGX_OK as isize { + return Err(Disconnected); + } + return Ok(()); + } + BufferSendResult::Error => return Err(Disconnected), + } + /*if first.start == first.end { + first.start = 0; + first.end = 0; // reuse the space in this buffer + break; // buffer empty + }*/ + if first.start as usize == first.data.len() { + self.buffers.pop_front(); + } + } + if (*(*conn).write).active() != 0 { + if let Some(del) = ngx_event_actions.del { + del((*conn).write, NGX_RS_WRITE_EVENT as isize, 0); + } + } + } + Ok(()) + } +} + +enum BufferSendResult { + Ok, + EndOfStream, + Again, + Error, +} + +impl WriteBuffer { + fn new() -> Self { + let data = vec![0; 16384]; + Self { + data: data.into_boxed_slice(), + start: 0, + end: 0, + } + } + + fn push<'a>(&mut self, data: &'a [u8]) -> &'a [u8] { + if (self.end as usize) < self.data.len() { + let available = self.data.len() - self.end as usize; + let to_copy = available.min(data.len()); + let next_end = self.end as usize + to_copy; + self.data[self.end as usize..next_end].copy_from_slice(&data[..to_copy]); + self.end = next_end as u32; + &data[to_copy..] + } else { + data + } + } + + unsafe fn send( + &mut self, + conn: *mut ngx_connection_t, + send_fn: unsafe extern "C" fn(*mut ngx_connection_t, *mut u8, usize) -> isize, + ) -> BufferSendResult { + let sendable = &self.data[self.start as usize..self.end as usize]; + if !sendable.is_empty() { + let result = send_fn(conn, sendable.as_ptr().cast_mut(), sendable.len()); + if result > 0 { + self.start += result as u32; + BufferSendResult::Ok + } else if result == 0 { + BufferSendResult::EndOfStream + } else if result == NGX_AGAIN as isize { + BufferSendResult::Again + } else { + BufferSendResult::Error + } + } else { + BufferSendResult::Ok + } + } +} + +unsafe extern "C" fn empty_get(_pc: *mut ngx_peer_connection_t, _data: *mut c_void) -> isize { + NGX_OK as isize +} + +unsafe extern "C" fn on_free(_pc: *mut ngx_peer_connection_t, _data: *mut c_void, _state: usize) {} + +unsafe extern "C" fn on_write(wev: *mut ngx_event_t) { + let conn = &mut *((*wev).data as *mut ngx_connection_t); + if conn.destroyed() != 0 { + return; + } + + if (*wev).timedout() != 0 { + return; + } + + if (*wev).timer_set() != 0 { + ngx_event_del_timer(wev); + } + + let conn_data = conn_data(conn); + if !conn_data.is_null() { + (*conn_data).send(); + } +} + +unsafe extern "C" fn on_read(rev: *mut ngx_event_t) { + let conn = &mut *((*rev).data as *mut ngx_connection_t); + if conn.destroyed() != 0 { + return; + } + + if (*rev).timedout() != 0 { + conn.set_timedout(1); + return; + } + + if (*rev).timer_set() != 0 { + ngx_event_del_timer(rev); + } + + let data = conn_data(conn); + if !data.is_null() { + if let Some(recv) = conn.recv { + let mut buf = [0; 16384]; + loop { + let result = recv(conn, buf.as_mut_ptr(), buf.len()); + if result > 0 { + let mut handshake_done = false; + let mut is_connected = false; + { + let mut state = match (*data).state.try_borrow_mut() { + Ok(state) => state, + Err(error) => { + log_borrow_mut_error("on_read: state borrow", error); + break; + } + }; + + match &mut *state { + State::WaitServerHandshake { conn, buffers } => { + let handshake_result = match (*data).check_handshake.try_borrow_mut() + { + Ok(mut check_handshake) => { + (check_handshake)(&buf[..result as usize]) + } + Err(error) => { + log_borrow_mut_error( + "on_read: check_handshake borrow", + error, + ); + Err(anyhow::anyhow!("Borrow error")) + } + }; + + match handshake_result { + Ok(_) => { + let new_conn = *conn; + *conn = std::ptr::null_mut(); + handshake_done = true; + *state = State::Connected { + conn: new_conn, + buffers: std::mem::take(buffers), + } + } + Err(_) => { + *state = State::Disconnected { + event: (*data).create_and_schedule_reconnect(), + reconnect_timeout: MIN_TIMEOUT_MS, + } + } + } + } + State::Connected { .. } => { + is_connected = true; + } + + State::Disconnected { .. } => {} + } + } // state not borrowed here any more + if handshake_done { + let back_data = match (*data).after_handshake.try_borrow_mut() { + Ok(mut after_handshake) => (after_handshake)(), + Err(error) => { + log_borrow_mut_error("on_read: after_handshake borrow", error); + Vec::new() + } + }; + if !back_data.is_empty() && (*data).write(&back_data).is_err() { + match (*data).state.try_borrow_mut() { + Ok(mut state) => { + *state = State::Disconnected { + event: (*data).create_and_schedule_reconnect(), + reconnect_timeout: MIN_TIMEOUT_MS, + }; + } + Err(error) => { + log_borrow_mut_error( + "on_read: state borrow (after_handshake)", + error, + ); + } + } + } + } + if is_connected { + let back_data = match (*data).on_read.try_borrow_mut() { + Ok(mut on_read) => (on_read)(&buf[..result as usize]), + Err(error) => { + log_borrow_mut_error("on_read: on_read borrow", error); + Vec::new() + } + }; + + if !back_data.is_empty() && (*data).write(&back_data).is_err() { + match (*data).state.try_borrow_mut() { + Ok(mut state) => { + *state = State::Disconnected { + event: (*data).create_and_schedule_reconnect(), + reconnect_timeout: MIN_TIMEOUT_MS, + }; + } + Err(error) => { + log_borrow_mut_error("on_read: state borrow (on_read)", error); + } + } + } + } + } else if result == 0 { + // error, signal connection close + match (*data).state.try_borrow_mut() { + Ok(mut state) => { + *state = State::Disconnected { + event: (*data).create_and_schedule_reconnect(), + reconnect_timeout: MIN_TIMEOUT_MS, + }; + } + Err(error) => { + log_borrow_mut_error("on_read: state borrow (result=0)", error); + } + } + break; + } else if result == NGX_AGAIN as isize { + if ngx_handle_read_event(conn.read, 0) != NGX_OK as isize { + match (*data).state.try_borrow_mut() { + Ok(mut state) => { + *state = State::Disconnected { + event: (*data).create_and_schedule_reconnect(), + reconnect_timeout: MIN_TIMEOUT_MS, + }; + } + Err(error) => { + log_borrow_mut_error("on_read: state borrow (NGX_AGAIN)", error); + } + } + } + break; + } else { + // error, retry reconnect + match (*data).state.try_borrow_mut() { + Ok(mut state) => { + *state = State::Disconnected { + event: (*data).create_and_schedule_reconnect(), + reconnect_timeout: MIN_TIMEOUT_MS, + }; + } + Err(error) => { + log_borrow_mut_error("on_read: state borrow (error)", error); + } + } + break; + } + } + } + } +} + +unsafe extern "C" fn on_reconnect_timeout(ev: *mut ngx_event_t) { + // Add null pointer checks + if ev.is_null() { + return; + } + + let data = (*ev).data as *const Inner; + if data.is_null() { + return; + } + + let data = &*data; + + // Use try_borrow_mut to avoid panic if already borrowed + let mut state = match data.state.try_borrow_mut() { + Ok(state) => state, + Err(error) => { + log_borrow_mut_error("on_reconnect_timeout: state borrow", error); + // State is already borrowed, schedule retry + if (*ev).timer_set() == 0 && ngx_quit == 0 && ngx_exiting == 0 && ngx_terminate == 0 { + ngx_event_add_timer(ev, MIN_TIMEOUT_MS); + } + return; + } + }; + + if let State::Disconnected { + reconnect_timeout, .. + } = &mut *state + { + if *reconnect_timeout < TIMEOUT_MS { + *reconnect_timeout = (*reconnect_timeout) * 2; + } + + if let Some(conn) = State::try_connect( + &data.path, + &data.name, + &*data.dummy_log as *const ngx_log_t as *mut ngx_log_t, + ) { + (*conn).data = (*ev).data; + (*(*conn).write).handler = Some(on_write); + (*(*conn).read).handler = Some(on_read); + + let mut buffers = WriteBuffers::default(); + buffers.push(&data.handshake_msg); + *state = match buffers.send(conn) { + Ok(()) => State::WaitServerHandshake { conn, buffers }, + Err(_) => { + let mut new_ev: Box = + Box::new(MaybeUninit::zeroed().assume_init()); + new_ev.handler = Some(on_reconnect_timeout); + new_ev.log = (*ngx_cycle).log; + new_ev.data = (*ev).data; + + if (*new_ev).timer_set() == 0 + && ngx_quit == 0 + && ngx_exiting == 0 + && ngx_terminate == 0 + { + ngx_event_add_timer(&mut *new_ev, *reconnect_timeout); + } + + State::Disconnected { + event: new_ev, + reconnect_timeout: *reconnect_timeout, + } + } + }; + } else if (*ev).timer_set() == 0 && ngx_quit == 0 && ngx_exiting == 0 && ngx_terminate == 0 + { + (*ev).handler = Some(on_reconnect_timeout); + ngx_event_add_timer(ev, *reconnect_timeout); + } + } +} + +unsafe extern "C" fn dummy_log_fn(_log: *mut ngx_log_t, _level: usize, _buf: *mut u8, _len: usize) { +} From cd7667a07ce3040664c88e8af3d51481b130b0b2 Mon Sep 17 00:00:00 2001 From: Ruslan Pislari Date: Thu, 9 Jul 2026 11:12:56 +0300 Subject: [PATCH 2/4] Potential fix for pull request finding fix Inner::send only flushes pending write buffers when the state is Connected Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- nginx_module/src/yamux_unix_socket.rs | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/nginx_module/src/yamux_unix_socket.rs b/nginx_module/src/yamux_unix_socket.rs index 8103029..62b76ef 100644 --- a/nginx_module/src/yamux_unix_socket.rs +++ b/nginx_module/src/yamux_unix_socket.rs @@ -353,13 +353,17 @@ impl Inner { } // Already borrowed, skip this send attempt }; - if let State::Connected { conn, buffers } = &mut *state { - if let Err(Disconnected) = buffers.send(*conn) { - *state = State::Disconnected { - event: self.create_and_schedule_reconnect(), - reconnect_timeout: MIN_TIMEOUT_MS, - }; + match &mut *state { + State::Connected { conn, buffers } + | State::WaitServerHandshake { conn, buffers } => { + if let Err(Disconnected) = buffers.send(*conn) { + *state = State::Disconnected { + event: self.create_and_schedule_reconnect(), + reconnect_timeout: MIN_TIMEOUT_MS, + }; + } } + State::Disconnected { .. } => {} } } From 3a75f1b702c2a1bed205720100c2d92c23a60702 Mon Sep 17 00:00:00 2001 From: Ruslan Pislari Date: Thu, 9 Jul 2026 11:13:14 +0300 Subject: [PATCH 3/4] Potential fix for pull request finding fix the reconnect backoff update can overshoot TIMEOUT_MS Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- nginx_module/src/yamux_unix_socket.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/nginx_module/src/yamux_unix_socket.rs b/nginx_module/src/yamux_unix_socket.rs index 62b76ef..382431f 100644 --- a/nginx_module/src/yamux_unix_socket.rs +++ b/nginx_module/src/yamux_unix_socket.rs @@ -722,7 +722,7 @@ unsafe extern "C" fn on_reconnect_timeout(ev: *mut ngx_event_t) { } = &mut *state { if *reconnect_timeout < TIMEOUT_MS { - *reconnect_timeout = (*reconnect_timeout) * 2; + *reconnect_timeout = (*reconnect_timeout * 2).min(TIMEOUT_MS); } if let Some(conn) = State::try_connect( From 23a4f6a0de2b4d8bb4eea9c557aa29e11a7b3a50 Mon Sep 17 00:00:00 2001 From: Ruslan Pislari Date: Thu, 9 Jul 2026 11:13:34 +0300 Subject: [PATCH 4/4] Potential fix for pull request finding sockaddr_un::sun_path is a fixed-size C string buffer that should remain NUL-terminated Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- nginx_module/src/yamux_unix_socket.rs | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/nginx_module/src/yamux_unix_socket.rs b/nginx_module/src/yamux_unix_socket.rs index 382431f..d96580d 100644 --- a/nginx_module/src/yamux_unix_socket.rs +++ b/nginx_module/src/yamux_unix_socket.rs @@ -215,8 +215,8 @@ impl State { let path = Path::new(path); let mut len = path.as_os_str().len(); - if len > sockaddr.sun_path.len() { - len = sockaddr.sun_path.len(); + if len >= sockaddr.sun_path.len() { + len = sockaddr.sun_path.len() - 1; // TODO: report warning } @@ -225,8 +225,9 @@ impl State { path.as_os_str().as_bytes().as_ptr(), sockaddr.sun_path.as_mut_ptr().cast(), len, - ) - }; + ); + } + sockaddr.sun_path[len] = 0; let mut name = NgxStr::from(name);