diff --git a/crates/dbx-core/src/db/http_tunnel.rs b/crates/dbx-core/src/db/http_tunnel.rs index 8d8d83df77..4e6132e8f0 100644 --- a/crates/dbx-core/src/db/http_tunnel.rs +++ b/crates/dbx-core/src/db/http_tunnel.rs @@ -301,9 +301,11 @@ fn effective_connect_timeout_secs(value: u64) -> u64 { #[cfg(test)] mod tests { use super::{script_url, validate_script_url, HttpTunnelManager}; + use std::sync::Arc; use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader}; use tokio::net::{TcpListener, TcpStream}; - use tokio::sync::Mutex; + use tokio::sync::{Mutex, Notify}; + use tokio::time::{timeout, Duration}; #[test] fn script_url_preserves_existing_query_and_appends_action() { @@ -331,16 +333,28 @@ mod tests { } #[tokio::test] - async fn manager_forwards_bytes_through_http_script_protocol() { + async fn manager_forwards_sequential_round_trips_and_closes_session() { let script = MockScript::start().await; let manager = HttpTunnelManager::new(); let local_port = manager.start_tunnel("test", &script.url, "secret", 5, "mysql.internal", 3306).await.unwrap(); let mut client = TcpStream::connect(("127.0.0.1", local_port)).await.unwrap(); - client.write_all(b"ping").await.unwrap(); - let mut response = [0_u8; 4]; - client.read_exact(&mut response).await.unwrap(); + let close_notification = script.closed.notified(); - assert_eq!(&response, b"ping"); + timeout(Duration::from_secs(10), async { + for index in 0..100_u32 { + let payload = format!("request-{index:03}-{}", "x".repeat((index % 17) as usize)); + client.write_all(payload.as_bytes()).await.unwrap(); + let mut response = vec![0_u8; payload.len()]; + client.read_exact(&mut response).await.unwrap(); + + assert_eq!(response, payload.as_bytes()); + } + }) + .await + .expect("100 sequential HTTP tunnel round trips timed out"); + + drop(client); + timeout(Duration::from_secs(2), close_notification).await.expect("HTTP tunnel session was not closed"); manager.stop_tunnel("test").await; script.handle.abort(); @@ -348,6 +362,7 @@ mod tests { struct MockScript { url: String, + closed: Arc, handle: tokio::task::JoinHandle<()>, } @@ -355,26 +370,29 @@ mod tests { async fn start() -> Self { let listener = TcpListener::bind(("127.0.0.1", 0)).await.unwrap(); let addr = listener.local_addr().unwrap(); - let buffered = std::sync::Arc::new(Mutex::new(Vec::::new())); + let buffered = Arc::new(Mutex::new(Vec::::new())); + let closed = Arc::new(Notify::new()); let handle = { let buffered = buffered.clone(); + let closed = closed.clone(); tokio::spawn(async move { loop { let Ok((stream, _)) = listener.accept().await else { break; }; let buffered = buffered.clone(); + let closed = closed.clone(); tokio::spawn(async move { - handle_mock_http_request(stream, buffered).await; + handle_mock_http_request(stream, buffered, closed).await; }); } }) }; - Self { url: format!("http://{addr}/dbx_tunnel.php"), handle } + Self { url: format!("http://{addr}/dbx_tunnel.php"), closed, handle } } } - async fn handle_mock_http_request(stream: TcpStream, buffered: std::sync::Arc>>) { + async fn handle_mock_http_request(stream: TcpStream, buffered: Arc>>, closed: Arc) { let mut reader = BufReader::new(stream); let mut request_line = String::new(); reader.read_line(&mut request_line).await.unwrap(); @@ -407,6 +425,9 @@ mod tests { } else { write_http_response(&mut stream, "200 OK", &bytes).await; } + } else if request_line.contains("dbx_action=close") { + write_http_response(&mut stream, "200 OK", b"OK").await; + closed.notify_one(); } else { write_http_response(&mut stream, "200 OK", b"OK").await; } diff --git a/deploy/dbx_tunnel.php b/deploy/dbx_tunnel.php index 8d6192424b..69706e7460 100644 --- a/deploy/dbx_tunnel.php +++ b/deploy/dbx_tunnel.php @@ -21,6 +21,18 @@ $DBX_TUNNEL_DIR = getenv('DBX_TUNNEL_DIR') ?: sys_get_temp_dir() . DIRECTORY_SEP $DBX_TUNNEL_ALLOWED_HOSTS = array_values(array_filter(array_map('trim', explode(',', getenv('DBX_TUNNEL_ALLOWED_HOSTS') ?: '')))); $DBX_TUNNEL_MAX_SESSION_SECONDS = max(30, (int) (getenv('DBX_TUNNEL_MAX_SESSION_SECONDS') ?: '3600')); +// Keep sequential protocol exchanges responsive, then restore the original idle cadence. +const DBX_WORKER_ACTIVE_POLL_US = 10000; +const DBX_WORKER_WARM_POLL_US = 50000; +const DBX_WORKER_IDLE_POLL_US = 200000; +const DBX_WORKER_ACTIVE_POLL_COUNT = 100; +const DBX_WORKER_WARM_POLL_COUNT = 20; + +// Standalone tests load the worker helpers without dispatching an HTTP request. +if (defined('DBX_TUNNEL_FUNCTIONS_ONLY') && DBX_TUNNEL_FUNCTIONS_ONLY) { + return; +} + if (PHP_SAPI === 'cli' && isset($argv[1]) && $argv[1] === '--dbx-worker') { run_worker($argv[2] ?? '', $argv[3] ?? '', (int) ($argv[4] ?? 0), (int) ($argv[5] ?? 10)); exit; @@ -160,6 +172,7 @@ function run_worker(string $dir, string $host, int $port, int $connectTimeout): stream_set_blocking($socket, false); $expiresAt = time() + $DBX_TUNNEL_MAX_SESSION_SECONDS; $lastActivity = time(); + $idlePolls = 0; try { while (time() < $expiresAt) { @@ -167,16 +180,18 @@ function run_worker(string $dir, string $host, int $port, int $connectTimeout): break; } + $activity = false; $inbound = drain_chunks($dir, 'in'); if ($inbound !== '') { write_all($socket, $inbound); $lastActivity = time(); + $activity = true; } $read = [$socket]; $write = []; $except = []; - $ready = @stream_select($read, $write, $except, 0, 200000); + $ready = @stream_select($read, $write, $except, 0, worker_poll_timeout_us($idlePolls)); if ($ready === false) { write_error($dir, 'Failed to poll target database socket'); break; @@ -194,9 +209,12 @@ function run_worker(string $dir, string $host, int $port, int $connectTimeout): } else { append_chunk($dir, 'out', $data); $lastActivity = time(); + $activity = true; } } + $idlePolls = next_worker_idle_poll_count($idlePolls, $activity); + if (time() - $lastActivity > $DBX_TUNNEL_MAX_SESSION_SECONDS) { break; } @@ -209,6 +227,25 @@ function run_worker(string $dir, string $host, int $port, int $connectTimeout): mark_closed($dir); } +function worker_poll_timeout_us(int $idlePolls): int +{ + if ($idlePolls < DBX_WORKER_ACTIVE_POLL_COUNT) { + return DBX_WORKER_ACTIVE_POLL_US; + } + if ($idlePolls < DBX_WORKER_ACTIVE_POLL_COUNT + DBX_WORKER_WARM_POLL_COUNT) { + return DBX_WORKER_WARM_POLL_US; + } + return DBX_WORKER_IDLE_POLL_US; +} + +function next_worker_idle_poll_count(int $idlePolls, bool $activity): int +{ + if ($activity) { + return 0; + } + return min($idlePolls + 1, DBX_WORKER_ACTIVE_POLL_COUNT + DBX_WORKER_WARM_POLL_COUNT); +} + function write_all($socket, string $data): void { $offset = 0; diff --git a/deploy/tests/http_tunnel_worker_test.php b/deploy/tests/http_tunnel_worker_test.php new file mode 100644 index 0000000000..a273932a48 --- /dev/null +++ b/deploy/tests/http_tunnel_worker_test.php @@ -0,0 +1,164 @@ + ['pipe', 'r'], + 1 => ['pipe', 'w'], + 2 => ['pipe', 'w'], + ], $pipes); + if (!is_resource($worker)) { + throw new RuntimeException('Failed to start tunnel worker'); + } + fclose($pipes[0]); + stream_set_blocking($pipes[1], false); + stream_set_blocking($pipes[2], false); + + $socket = @stream_socket_accept($server, 5); + if ($socket === false) { + throw new RuntimeException('Tunnel worker did not connect to echo target'); + } + stream_set_blocking($socket, false); + + $roundTrips = 100; + $startedAt = microtime(true); + for ($index = 0; $index < $roundTrips; $index++) { + $payload = pack('N', $index) . hash('sha256', (string) $index, true); + append_chunk($sessionDir, 'in', $payload); + + $forwarded = read_exact_with_timeout($socket, strlen($payload), 2.0); + assert_same($payload, $forwarded, 'TCP target receives ordered payload ' . $index); + write_all($socket, $forwarded); + + $reply = drain_exact_with_timeout($sessionDir, 'out', strlen($payload), 2.0); + assert_same($payload, $reply, 'tunnel returns ordered payload ' . $index); + } + $elapsed = microtime(true) - $startedAt; + if ($elapsed >= 10.0) { + throw new RuntimeException(sprintf( + 'Sequential tunnel round trips took %.3fs; the old fixed wait pattern takes about 20s', + $elapsed + )); + } + + touch($sessionDir . DIRECTORY_SEPARATOR . 'close'); + wait_for_path($sessionDir . DIRECTORY_SEPARATOR . 'closed', 3.0); + + printf("ok - %d sequential round trips in %.3fs; ordered payloads and clean shutdown verified\n", $roundTrips, $elapsed); +} finally { + if (is_resource($socket)) { + fclose($socket); + } + if (is_resource($server)) { + fclose($server); + } + if (is_resource($worker)) { + foreach ([1, 2] as $pipeIndex) { + if (isset($pipes[$pipeIndex]) && is_resource($pipes[$pipeIndex])) { + fclose($pipes[$pipeIndex]); + } + } + $status = proc_get_status($worker); + if ($status['running']) { + proc_terminate($worker); + } + proc_close($worker); + } + if (is_dir($baseDir)) { + remove_dir($baseDir); + } +} + +function read_exact_with_timeout($socket, int $length, float $timeoutSeconds): string +{ + $data = ''; + $deadline = microtime(true) + $timeoutSeconds; + while (strlen($data) < $length) { + $chunk = fread($socket, $length - strlen($data)); + if ($chunk === false) { + throw new RuntimeException('Failed to read echo target socket'); + } + if ($chunk !== '') { + $data .= $chunk; + continue; + } + if (microtime(true) >= $deadline) { + throw new RuntimeException('Timed out reading echo target socket'); + } + usleep(1000); + } + return $data; +} + +function drain_exact_with_timeout(string $dir, string $name, int $length, float $timeoutSeconds): string +{ + $data = ''; + $deadline = microtime(true) + $timeoutSeconds; + while (strlen($data) < $length) { + $data .= drain_chunks($dir, $name); + if (strlen($data) >= $length) { + break; + } + if (microtime(true) >= $deadline) { + throw new RuntimeException('Timed out draining tunnel queue'); + } + usleep(1000); + } + return $data; +} + +function wait_for_path(string $path, float $timeoutSeconds): void +{ + $deadline = microtime(true) + $timeoutSeconds; + while (!is_file($path)) { + if (microtime(true) >= $deadline) { + throw new RuntimeException('Timed out waiting for ' . $path); + } + usleep(10000); + } +} + +function assert_same($expected, $actual, string $message): void +{ + if ($expected !== $actual) { + throw new RuntimeException($message); + } +}