hyperdb_api/process.rs
1// Copyright (c) 2026, Salesforce, Inc. All rights reserved.
2// SPDX-License-Identifier: Apache-2.0 OR MIT
3
4//! Hyper server process management.
5//!
6//! This module provides [`HyperProcess`] for spawning and managing local hyperd server instances.
7//!
8//! # Callback Connection Architecture
9//!
10//! The `HyperProcess` uses a **callback connection** mechanism for reliable process lifecycle
11//! management. This works as follows:
12//!
13//! 1. **Startup**: The client creates a TCP listener on an ephemeral port (the "callback proxy")
14//! and passes this address to hyperd via `--callback-connection`. Hyper connects back to this
15//! listener and sends its actual listen endpoint over this connection.
16//!
17//! 2. **Runtime**: The callback connection remains open for the lifetime of the `HyperProcess`.
18//! This connection acts as a "dead man's switch" - Hyper monitors it continuously.
19//!
20//! 3. **Graceful Shutdown**: When `HyperProcess` is dropped or explicitly shut down, the callback
21//! connection is closed. Hyper detects this and initiates graceful shutdown automatically.
22//!
23//! 4. **Crash Safety**: If the client process crashes or is killed, the OS automatically closes
24//! the TCP connection. Hyper detects this and shuts down gracefully, preventing orphan
25//! processes. This is the key advantage over signal-based shutdown.
26//!
27//! # Protocol Details
28//!
29//! The callback connection protocol is simple:
30//! - Client listens on `127.0.0.1:<ephemeral_port>`
31//! - Client starts hyperd with `--callback-connection=tab.tcp://127.0.0.1:<port>`
32//! - Hyper connects to this address and sends: `[1 byte length][N bytes descriptor]`
33//! - The descriptor is the actual listen endpoint (e.g., `tab.tcp://localhost:54321`)
34//! - Connection stays open until shutdown is desired
35//!
36//! # Listen Modes
37//!
38//! `HyperProcess` supports different listen modes via [`ListenMode`]:
39//!
40//! - **`LibPq`** (default): `PostgreSQL` wire protocol for full read/write access
41//! - **Grpc**: gRPC protocol for query-only Arrow-based access
42//! - **Both**: Both protocols enabled (libpq for read/write, gRPC for Arrow queries)
43//!
44//! ```no_run
45//! use hyperdb_api::{HyperProcess, ListenMode, Parameters, Result};
46//!
47//! fn main() -> Result<()> {
48//! // gRPC only
49//! let mut params = Parameters::new();
50//! params.set_listen_mode(ListenMode::Grpc { port: 0 });
51//! let hyper = HyperProcess::new(None, Some(¶ms))?;
52//! println!("gRPC endpoint: {}", hyper.grpc_endpoint().unwrap());
53//!
54//! // Both libpq and gRPC
55//! let mut params = Parameters::new();
56//! params.set_listen_mode(ListenMode::Both { grpc_port: 7484 });
57//! let hyper = HyperProcess::new(None, Some(¶ms))?;
58//! println!("libpq endpoint: {}", hyper.endpoint().unwrap());
59//! println!("gRPC endpoint: {}", hyper.grpc_endpoint().unwrap());
60//! Ok(())
61//! }
62//! ```
63
64use std::io::Read;
65use std::net::{Shutdown, TcpListener, TcpStream};
66use std::path::{Path, PathBuf};
67use std::process::{Child, Command, Stdio};
68use std::sync::Arc;
69use std::sync::atomic::{AtomicBool, Ordering};
70use std::thread;
71use std::time::Duration;
72
73#[cfg(unix)]
74use std::os::unix::process::CommandExt;
75
76#[cfg(any(unix, windows))]
77use hyperdb_api_core::client::ConnectionEndpoint;
78
79use tracing::info;
80
81use crate::error::{Error, Result};
82
83/// Specifies which protocols `HyperProcess` should listen on.
84///
85/// # Examples
86///
87/// ```
88/// use hyperdb_api::{ListenMode, Parameters};
89///
90/// // LibPq only (default) - for full read/write access
91/// let mut params = Parameters::new();
92/// params.set_listen_mode(ListenMode::LibPq);
93///
94/// // gRPC only - for Arrow-based query access
95/// let mut params = Parameters::new();
96/// params.set_listen_mode(ListenMode::Grpc { port: 0 }); // auto-assign port
97///
98/// // Both protocols - libpq for writes, gRPC for Arrow queries
99/// let mut params = Parameters::new();
100/// params.set_listen_mode(ListenMode::Both { grpc_port: 7484 });
101/// ```
102#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
103pub enum ListenMode {
104 /// `PostgreSQL` wire protocol only (default).
105 ///
106 /// This is the traditional connection mode that supports all Hyper features
107 /// including read and write operations.
108 #[default]
109 LibPq,
110
111 /// gRPC protocol only.
112 ///
113 /// This mode is optimized for query-only workloads and returns results in
114 /// Arrow IPC format. Note that gRPC mode does not support write operations.
115 ///
116 /// Set `port` to 0 to auto-assign an available port.
117 Grpc {
118 /// The port to listen on (0 for auto-assign).
119 port: u16,
120 },
121
122 /// Both libpq and gRPC protocols.
123 ///
124 /// This mode enables full read/write access via libpq while also providing
125 /// gRPC access for Arrow-based queries. The libpq port is auto-assigned,
126 /// while the gRPC port is specified.
127 ///
128 /// Note: When using `Both` mode, the callback connection returns the libpq
129 /// endpoint. Use `HyperProcess::grpc_endpoint()` to get the gRPC endpoint.
130 Both {
131 /// The gRPC port to listen on (cannot be 0 - must be a specific port).
132 grpc_port: u16,
133 },
134}
135
136/// A running Hyper server instance.
137///
138/// This struct manages the lifecycle of a local Hyper server process using a callback
139/// connection for reliable shutdown. The server is automatically shut down when this
140/// object is dropped.
141///
142/// # Callback Connection (Dead Man's Switch)
143///
144/// Unlike traditional process management that relies on signals, `HyperProcess` maintains
145/// a TCP connection to the Hyper server. When this connection is closed (either explicitly
146/// or because the client process exits), Hyper automatically shuts down gracefully.
147///
148/// This provides several benefits:
149/// - **No orphan processes**: If your application crashes, Hyper shuts down automatically
150/// - **Graceful shutdown**: Hyper can flush data and clean up properly
151/// - **Cross-platform**: Works reliably on macOS, Linux, and Windows
152///
153/// # Example
154///
155/// ```no_run
156/// use hyperdb_api::{HyperProcess, Result};
157///
158/// fn main() -> Result<()> {
159/// // Start a Hyper server (auto-detect hyperd location)
160/// let hyper = HyperProcess::new(None, None)?;
161///
162/// println!("Hyper server running at: {}", hyper.endpoint().unwrap());
163///
164/// // Server automatically shuts down when `hyper` goes out of scope
165/// // via the callback connection mechanism
166/// Ok(())
167/// }
168/// ```
169#[must_use = "HyperProcess will shut down when dropped; store it to keep the server running"]
170#[derive(Debug)]
171pub struct HyperProcess {
172 /// The child process handle.
173 child: Option<Child>,
174 /// The libpq endpoint descriptor (host:port or socket path), if libpq is enabled.
175 endpoint: Option<String>,
176 /// The parsed connection endpoint for libpq.
177 connection_endpoint: Option<ConnectionEndpoint>,
178 /// The gRPC endpoint (host:port), if gRPC is enabled.
179 grpc_endpoint: Option<String>,
180 /// Path to the hyperd executable.
181 #[expect(
182 dead_code,
183 reason = "retained for diagnostics and future restart/respawn paths"
184 )]
185 hyperd_path: PathBuf,
186 /// Whether shutdown has been initiated.
187 shutdown_initiated: Arc<AtomicBool>,
188 /// The callback connection to hyperd.
189 /// Keeping this open maintains the "dead man's switch" - when dropped, Hyper shuts down.
190 callback_connection: Option<TcpStream>,
191 /// The listen mode this process was started with.
192 listen_mode: ListenMode,
193 /// The transport mode this process was started with.
194 transport_mode: TransportMode,
195 /// The socket directory for UDS connections (Unix only).
196 /// This directory is automatically cleaned up on drop.
197 #[cfg(unix)]
198 socket_directory: Option<PathBuf>,
199 /// The pipe name for Named Pipe connections (Windows only).
200 #[cfg(windows)]
201 pipe_name: Option<String>,
202 /// The log directory where hyperd writes its log files.
203 log_dir: Option<PathBuf>,
204}
205
206impl HyperProcess {
207 /// Starts a new Hyper server instance.
208 ///
209 /// This method:
210 /// 1. Creates a callback listener on an ephemeral port
211 /// 2. Starts the hyperd process with the callback connection address
212 /// 3. Waits for Hyper to connect back and provide its listen endpoint
213 ///
214 /// # Arguments
215 ///
216 /// * `hyper_path` - Optional path to the hyperd executable. If `None`, searches
217 /// in common locations (`HYPERD_PATH` env var, then known build output paths).
218 /// * `parameters` - Optional parameters for the server.
219 ///
220 /// # Errors
221 ///
222 /// Returns an error if:
223 /// - The hyperd executable cannot be found
224 /// - The callback listener cannot be created
225 /// - The server fails to start
226 /// - Hyper doesn't connect back within the timeout (30 seconds)
227 ///
228 /// # Example
229 ///
230 /// ```no_run
231 /// use hyperdb_api::{HyperProcess, Result};
232 /// use std::path::Path;
233 ///
234 /// fn main() -> Result<()> {
235 /// // Auto-detect hyperd location
236 /// let hyper = HyperProcess::new(None, None)?;
237 ///
238 /// // Or specify explicit path
239 /// let hyper2 = HyperProcess::new(
240 /// Some(Path::new("/path/to/hyperd")),
241 /// None,
242 /// )?;
243 /// Ok(())
244 /// }
245 /// ```
246 pub fn new(hyper_path: Option<&Path>, parameters: Option<&Parameters>) -> Result<Self> {
247 let hyperd_path = match hyper_path {
248 Some(path) => path.to_path_buf(),
249 None => Self::find_hyperd()?,
250 };
251 Self::start_server(&hyperd_path, parameters)
252 }
253
254 /// Resolves the hyperd executable from the `HYPERD_PATH` environment
255 /// variable. The value can point at the executable directly, or at a
256 /// directory containing it.
257 ///
258 /// If `HYPERD_PATH` is unset, returns an error instructing the caller
259 /// to either set it or run the `hyperdb-bootstrap` downloader to
260 /// install a pinned release at `.hyperd/current/hyperd`.
261 fn find_hyperd() -> Result<PathBuf> {
262 #[cfg(windows)]
263 const HYPERD_EXE: &str = "hyperd.exe";
264 #[cfg(not(windows))]
265 const HYPERD_EXE: &str = "hyperd";
266
267 let Ok(path_str) = std::env::var("HYPERD_PATH") else {
268 // Walk up from CWD looking for .hyperd/current/<exe> written by
269 // `hyperdb-bootstrap download`. This lets `node examples/foo.mjs`
270 // run from any subdirectory of the repo without exporting HYPERD_PATH.
271 if let Ok(cwd) = std::env::current_dir() {
272 let mut dir = cwd.as_path();
273 loop {
274 let candidate = dir.join(".hyperd").join("current").join(HYPERD_EXE);
275 if candidate.exists() {
276 return Ok(candidate);
277 }
278 match dir.parent() {
279 Some(parent) => dir = parent,
280 None => break,
281 }
282 }
283 }
284 return Err(Error::config(
285 "HYPERD_PATH is not set. Point it at a hyperd executable, \
286 or run `make download-hyperd` (or `cargo run -p hyperdb-bootstrap -- download`) \
287 to install a pinned release at `.hyperd/current/hyperd`.",
288 ));
289 };
290
291 let path = PathBuf::from(&path_str);
292 if path.is_dir() {
293 let with_exe = path.join(HYPERD_EXE);
294 if with_exe.exists() {
295 return Ok(with_exe);
296 }
297 #[cfg(windows)]
298 {
299 let without_exe = path.join("hyperd");
300 if without_exe.exists() {
301 return Ok(without_exe);
302 }
303 }
304 return Err(Error::config(format!(
305 "HYPERD_PATH set to '{path_str}' but {HYPERD_EXE} not found in that directory"
306 )));
307 }
308 if path.exists() {
309 return Ok(path);
310 }
311 #[cfg(windows)]
312 {
313 let with_ext = PathBuf::from(format!("{path_str}.exe"));
314 if with_ext.exists() {
315 return Ok(with_ext);
316 }
317 }
318 Err(Error::config(format!(
319 "HYPERD_PATH set to '{}' but hyperd executable not found (checked: {})",
320 path_str,
321 path.display()
322 )))
323 }
324
325 /// Starts the hyperd server process with callback connection.
326 fn start_server(hyperd_path: &Path, parameters: Option<&Parameters>) -> Result<Self> {
327 // Verify hyperd exists
328 if !hyperd_path.exists() {
329 return Err(Error::config(format!(
330 "Hyper executable not found at: {}",
331 hyperd_path.display()
332 )));
333 }
334
335 info!(
336 target: "hyperdb_api",
337 path = %hyperd_path.display(),
338 "hyperd-starting"
339 );
340
341 // Create callback listener on ephemeral port
342 // This is the "dead man's switch" - when this connection is closed, Hyper shuts down
343 let callback_listener = TcpListener::bind("127.0.0.1:0")
344 .map_err(|e| Error::connection_with_io("Failed to create callback listener", e))?;
345
346 let callback_port = callback_listener
347 .local_addr()
348 .map_err(|e| Error::connection_with_io("Failed to get callback port", e))?
349 .port();
350
351 // Set a timeout for accepting the callback connection
352 callback_listener.set_nonblocking(false).map_err(|e| {
353 Error::connection_with_io("Failed to set callback listener to blocking", e)
354 })?;
355
356 // Check if user wants to disable default parameters
357 let use_defaults = parameters.is_none_or(|p| !p.contains_key(NO_DEFAULT_PARAMETERS));
358
359 // Get the listen mode
360 let listen_mode = parameters.and_then(|p| p.listen_mode).unwrap_or_default();
361
362 // Get transport mode. The default is TCP on every platform, and that
363 // is now a measured choice rather than a pending one: IPC wins latency
364 // but loses bulk streaming. On macOS/UDS, connect is ~30% faster and a
365 // small round-trip ~19% faster, while streamed reads past ~10k rows
366 // cost up to +62%; Windows Named Pipes show the same shape (+34-41% on
367 // single-connection inserts, -76% on async full scans). TCP therefore
368 // remains the right default for mixed workloads, and is retained here
369 // pending the separate transport change.
370 //
371 // Figures and methodology: hyperdb-api-core/docs/IPC_IMPLEMENTATION.md
372 // and docs/BENCHMARK_GUIDE.md.
373 #[cfg(unix)]
374 let transport_mode = parameters
375 .and_then(|p| p.transport_mode)
376 .unwrap_or(TransportMode::Tcp);
377 #[cfg(windows)]
378 let transport_mode = parameters
379 .and_then(|p| p.transport_mode)
380 .unwrap_or(TransportMode::Tcp);
381 #[cfg(not(any(unix, windows)))]
382 let transport_mode = TransportMode::Tcp;
383
384 // Create socket directory for UDS if needed (Unix only)
385 #[cfg(unix)]
386 let socket_directory: Option<PathBuf> = if transport_mode == TransportMode::Ipc {
387 // Use custom directory if provided, otherwise create temp directory
388 let dir = if let Some(custom_dir) =
389 parameters.and_then(|p| p.domain_socket_directory.as_ref())
390 {
391 custom_dir.clone()
392 } else {
393 // Create a temp directory for the socket
394 let temp_dir = std::env::temp_dir().join(format!("hyper-{}", std::process::id()));
395 std::fs::create_dir_all(&temp_dir).map_err(|e| {
396 Error::connection_with_io("Failed to create socket directory", e)
397 })?;
398 temp_dir
399 };
400 Some(dir)
401 } else {
402 None
403 };
404
405 // On non-Unix platforms there is no UDS socket directory; the variable
406 // is only referenced inside `#[cfg(unix)]` blocks so we do not need a
407 // placeholder binding here.
408
409 // Create pipe name for Named Pipes if needed (Windows only)
410 #[cfg(windows)]
411 let pipe_name: Option<String> = if transport_mode == TransportMode::Ipc {
412 Some(format!("hyper-{}", std::process::id()))
413 } else {
414 None
415 };
416
417 // Build command arguments
418 let mut cmd = Command::new(hyperd_path);
419
420 // The "run" subcommand starts the server
421 cmd.arg("run");
422
423 // Callback connection - Hyper will connect to this and send its endpoint
424 // When this connection is closed, Hyper will shut down gracefully
425 cmd.arg("--callback-connection")
426 .arg(format!("tab.tcp://127.0.0.1:{callback_port}"));
427
428 // Configure listen connection based on mode and transport
429 // Connection string formats:
430 // - tab.tcp://host:port - libpq over TCP
431 // - tab.domain://<dir>/domain/<name> - libpq over Unix Domain Socket
432 // - tcp.grpc://host:port - gRPC
433 #[cfg(unix)]
434 let listen_connection = if transport_mode == TransportMode::Ipc {
435 let socket_dir = socket_directory.as_ref().unwrap();
436 match listen_mode {
437 ListenMode::LibPq => format!("tab.domain://{}/domain/hyper", socket_dir.display()),
438 ListenMode::Grpc { port } => format!("tcp.grpc://127.0.0.1:{port}"),
439 ListenMode::Both { grpc_port } => {
440 format!(
441 "tab.domain://{}/domain/hyper,tcp.grpc://127.0.0.1:{}",
442 socket_dir.display(),
443 grpc_port
444 )
445 }
446 }
447 } else {
448 match listen_mode {
449 ListenMode::LibPq => "tab.tcp://localhost:0".to_string(),
450 ListenMode::Grpc { port } => format!("tcp.grpc://127.0.0.1:{port}"),
451 ListenMode::Both { grpc_port } => {
452 format!("tab.tcp://localhost:0,tcp.grpc://127.0.0.1:{grpc_port}")
453 }
454 }
455 };
456
457 #[cfg(windows)]
458 let listen_connection = if transport_mode == TransportMode::Ipc {
459 let pname = pipe_name.as_ref().unwrap();
460 match listen_mode {
461 ListenMode::LibPq => format!("tab.pipe://./pipe/{pname}"),
462 ListenMode::Grpc { port } => format!("tcp.grpc://127.0.0.1:{port}"),
463 ListenMode::Both { grpc_port } => {
464 format!("tab.pipe://./pipe/{pname},tcp.grpc://127.0.0.1:{grpc_port}")
465 }
466 }
467 } else {
468 match listen_mode {
469 ListenMode::LibPq => "tab.tcp://localhost:0".to_string(),
470 ListenMode::Grpc { port } => format!("tcp.grpc://127.0.0.1:{port}"),
471 ListenMode::Both { grpc_port } => {
472 format!("tab.tcp://localhost:0,tcp.grpc://127.0.0.1:{grpc_port}")
473 }
474 }
475 };
476
477 #[cfg(not(any(unix, windows)))]
478 let listen_connection = match listen_mode {
479 ListenMode::LibPq => "tab.tcp://localhost:0".to_string(),
480 ListenMode::Grpc { port } => format!("tcp.grpc://127.0.0.1:{}", port),
481 ListenMode::Both { grpc_port } => {
482 format!("tab.tcp://localhost:0,tcp.grpc://127.0.0.1:{}", grpc_port)
483 }
484 };
485
486 cmd.arg("--listen-connection").arg(&listen_connection);
487
488 // Helper to check if a parameter is already set by the user
489 let user_has_param =
490 |key: &str| -> bool { parameters.is_some_and(|p| p.contains_key(key)) };
491
492 // Apply default instance parameters (matching C++ HyperProcess behavior)
493 // These can be overridden by user parameters or disabled entirely with NO_DEFAULT_PARAMETERS
494 if use_defaults {
495 // Initial user for the Hyper instance
496 if !user_has_param("init_user") {
497 cmd.arg("--init-user=tableau_internal_user");
498 }
499
500 // Enable gRPC threads if gRPC mode is enabled
501 // Required for gRPC to function - without this, Hyper will fail to start with:
502 // "gRPC threads are required for running gRPC services"
503 // Using 4 threads as a reasonable default (can be overridden by user)
504 if matches!(
505 listen_mode,
506 ListenMode::Grpc { .. } | ListenMode::Both { .. }
507 ) && !user_has_param("grpc_threads")
508 {
509 cmd.arg("--grpc-threads=4");
510 }
511
512 // Enable gRPC result persistence if gRPC mode is enabled
513 // This is required for ADAPTIVE and ASYNC transfer modes
514 if matches!(
515 listen_mode,
516 ListenMode::Grpc { .. } | ListenMode::Both { .. }
517 ) && !user_has_param("grpc_persist_results")
518 {
519 cmd.arg("--grpc-persist-results=true");
520 }
521
522 // Default language setting
523 if !user_has_param("language") {
524 cmd.arg("--language=en_US");
525 }
526
527 // Log configuration: file-based JSON logging
528 if !user_has_param("log_config") {
529 cmd.arg(format!("--log-config={DEFAULT_LOG_CONFIG}"));
530 }
531
532 // Date style for date parsing (Month-Day-Year)
533 if !user_has_param("date_style") {
534 cmd.arg("--date-style=MDY");
535 }
536
537 // Enforce strict date_style (day/month/year ordering must match exactly)
538 if !user_has_param("date_style_lenient") {
539 cmd.arg("--date-style-lenient=false");
540 }
541
542 // Set default log directory to current directory
543 if !user_has_param("log_dir")
544 && let Ok(cwd) = std::env::current_dir()
545 {
546 cmd.arg(format!("--log-dir={}", cwd.display()));
547 }
548
549 // Disable password requirement for local development
550 if !user_has_param("no_password") {
551 cmd.arg("--no-password");
552 }
553
554 // Skip license check for local development
555 if !user_has_param("skip_license") {
556 cmd.arg("--skip-license");
557 }
558
559 // Default new .hyper databases to file format version 3, which
560 // adds support for 128-bit NUMERICs (required to ingest parquet
561 // files whose decimal columns are stored as DECIMAL128).
562 // File format 3 has shipped since Hyper 2022.4.0.
563 if !user_has_param("default_database_version") {
564 cmd.arg("--default-database-version=3");
565 }
566 }
567
568 // Add custom parameters from user
569 if let Some(params) = parameters {
570 for (key, value) in params.iter() {
571 // Skip internal/special parameters
572 if key == "callback_connection"
573 || key == "listen_connection"
574 || key == NO_DEFAULT_PARAMETERS
575 {
576 continue;
577 }
578
579 // Convert underscores to dashes for command-line arguments
580 let cli_key = key.replace('_', "-");
581
582 if value.is_empty() {
583 cmd.arg(format!("--{cli_key}"));
584 } else {
585 cmd.arg(format!("--{cli_key}={value}"));
586 }
587 }
588 }
589
590 // Resolve the effective log directory for later access via log_dir()
591 let resolved_log_dir =
592 if let Some(user_dir) = parameters.and_then(|p| p.get("log_dir")).map(PathBuf::from) {
593 Some(user_dir)
594 } else if use_defaults {
595 std::env::current_dir().ok()
596 } else {
597 None
598 };
599
600 // Redirect stdout/stderr to null - we get the endpoint via callback connection
601 cmd.stdout(Stdio::null());
602 cmd.stderr(Stdio::null());
603
604 // On Unix, start hyperd in its own process group so it doesn't receive
605 // Ctrl-C (SIGINT) signals meant for the parent process. This allows the
606 // parent to handle Ctrl-C gracefully and properly shut down hyperd via
607 // the callback connection mechanism.
608 #[cfg(unix)]
609 cmd.process_group(0);
610
611 // On Windows, prevent a console window from flashing when spawning hyperd.
612 // CREATE_NO_WINDOW (0x08000000) suppresses the creation of a visible console.
613 #[cfg(windows)]
614 {
615 use std::os::windows::process::CommandExt;
616 const CREATE_NO_WINDOW: u32 = 0x08000000;
617 cmd.creation_flags(CREATE_NO_WINDOW);
618 }
619
620 // Start the process
621 let child = cmd.spawn().map_err(|e| {
622 Error::connection_with_io(
623 format!("Failed to start Hyper server at {}", hyperd_path.display()),
624 e,
625 )
626 })?;
627
628 // Wait for Hyper to connect back to our callback listener
629 let (callback_connection, callback_endpoint) = Self::wait_for_callback(&callback_listener)?;
630
631 // Determine the endpoints based on listen mode and callback response
632 let (endpoint, grpc_endpoint) = match listen_mode {
633 ListenMode::LibPq => {
634 // Callback returns libpq endpoint
635 (Some(callback_endpoint), None)
636 }
637 ListenMode::Grpc { port } => {
638 // Callback returns gRPC endpoint (with resolved port if auto-assigned)
639 let grpc_ep = if port == 0 {
640 // The callback returns the actual gRPC endpoint with resolved port
641 callback_endpoint
642 } else {
643 format!("127.0.0.1:{port}")
644 };
645 (None, Some(grpc_ep))
646 }
647 ListenMode::Both { grpc_port } => {
648 // Callback returns libpq endpoint, gRPC uses specified port
649 (
650 Some(callback_endpoint),
651 Some(format!("127.0.0.1:{grpc_port}")),
652 )
653 }
654 };
655
656 // Parse connection endpoint if we have a libpq endpoint
657 let connection_endpoint = endpoint.as_ref().map(|ep| {
658 #[cfg(unix)]
659 {
660 // Check if it's a UDS path (contains path separator but no colon with port)
661 if ep.starts_with('/') || socket_directory.is_some() {
662 // UDS endpoint - construct from socket directory
663 if let Some(ref dir) = socket_directory {
664 return ConnectionEndpoint::domain_socket(dir, "hyper");
665 }
666 // Parse as path
667 let path = std::path::Path::new(ep);
668 let dir = path.parent().unwrap_or(std::path::Path::new("/"));
669 let name = path.file_name().and_then(|n| n.to_str()).unwrap_or("hyper");
670 return ConnectionEndpoint::domain_socket(dir, name);
671 }
672 }
673 #[cfg(windows)]
674 {
675 // Check if it's a Named Pipe endpoint
676 if let Some(ref pname) = pipe_name {
677 return ConnectionEndpoint::named_pipe(".", pname);
678 }
679 }
680 // TCP endpoint (host:port format)
681 let parts: Vec<&str> = ep.split(':').collect();
682 if parts.len() == 2
683 && let Ok(port) = parts[1].parse::<u16>()
684 {
685 return ConnectionEndpoint::tcp(parts[0], port);
686 }
687 ConnectionEndpoint::tcp("localhost", 7483) // fallback
688 });
689
690 Ok(HyperProcess {
691 child: Some(child),
692 endpoint,
693 connection_endpoint,
694 grpc_endpoint,
695 hyperd_path: hyperd_path.to_path_buf(),
696 shutdown_initiated: Arc::new(AtomicBool::new(false)),
697 callback_connection: Some(callback_connection),
698 listen_mode,
699 transport_mode,
700 log_dir: resolved_log_dir,
701 #[cfg(unix)]
702 socket_directory,
703 #[cfg(windows)]
704 pipe_name,
705 })
706 }
707
708 /// Waits for Hyper to connect to our callback listener and send its endpoint.
709 ///
710 /// Protocol:
711 /// - Hyper connects to the callback listener
712 /// - Hyper sends: [1 byte length][N bytes connection descriptor string]
713 /// - Connection descriptor format: "tab.tcp://host:port"
714 fn wait_for_callback(listener: &TcpListener) -> Result<(TcpStream, String)> {
715 // Set a timeout for accepting connections
716 listener.set_nonblocking(true).ok();
717
718 let timeout = Duration::from_secs(60);
719 let start = std::time::Instant::now();
720
721 // Poll for incoming connection with timeout
722 let mut stream = loop {
723 if start.elapsed() > timeout {
724 return Err(Error::internal(
725 "Timeout waiting for Hyper to connect to callback listener. \
726 Hyper may have failed to start - check hyperd logs for details.",
727 ));
728 }
729
730 match listener.accept() {
731 Ok((stream, _addr)) => break stream,
732 Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {
733 thread::sleep(Duration::from_millis(50));
734 }
735 Err(e) => {
736 return Err(Error::connection_with_io(
737 "Failed to accept callback connection",
738 e,
739 ));
740 }
741 }
742 };
743
744 // Set stream back to blocking for reading
745 stream.set_nonblocking(false).map_err(|e| {
746 Error::connection_with_io("Failed to set callback stream to blocking", e)
747 })?;
748
749 // Set read timeout
750 stream.set_read_timeout(Some(Duration::from_secs(10))).ok();
751
752 // Read the endpoint descriptor from Hyper
753 // Protocol: [1 byte length][N bytes descriptor string]
754 let mut len_buf = [0u8; 1];
755 stream.read_exact(&mut len_buf).map_err(|e| {
756 Error::connection_with_io("Failed to read endpoint length from Hyper", e)
757 })?;
758
759 let len = len_buf[0] as usize;
760 if len == 0 {
761 return Err(Error::internal("Hyper sent empty endpoint descriptor"));
762 }
763
764 let mut descriptor_buf = vec![0u8; len];
765 stream.read_exact(&mut descriptor_buf).map_err(|e| {
766 Error::connection_with_io("Failed to read endpoint descriptor from Hyper", e)
767 })?;
768
769 let descriptor = String::from_utf8(descriptor_buf)
770 .map_err(|e| Error::internal(format!("Invalid UTF-8 in endpoint descriptor: {e}")))?;
771
772 // Trim null bytes and whitespace that Hyper may include
773 let descriptor = descriptor.trim_matches(|c: char| c == '\0' || c.is_whitespace());
774
775 // Parse the connection descriptor (format: "tab.tcp://host:port")
776 let endpoint = Self::parse_connection_descriptor(descriptor)?;
777
778 // Clear read timeout for the connection we'll keep open
779 stream.set_read_timeout(None).ok();
780
781 info!(
782 target: "hyperdb_api",
783 %endpoint,
784 "hyperd-started"
785 );
786
787 Ok((stream, endpoint))
788 }
789
790 /// Parses a connection descriptor to extract host:port, socket path, or pipe path.
791 ///
792 /// Input formats:
793 /// - "tab.tcp://host:port" → "host:port"
794 /// - "tab.domain://<dir>/domain/<name>" → "<dir>/domain/<name>" (socket path)
795 /// - "tab.pipe://<host>/pipe/<name>" → "<host>/pipe/<name>" (named pipe)
796 /// - "tcp.grpc://host:port" → "host:port"
797 fn parse_connection_descriptor(descriptor: &str) -> Result<String> {
798 // Handle domain socket format
799 if let Some(rest) = descriptor.strip_prefix("tab.domain://") {
800 // Return the full path for UDS
801 if let Some(idx) = rest.find("/domain/") {
802 let dir = &rest[..idx];
803 let name = &rest[idx + 8..]; // "/domain/".len() == 8
804 let socket_path = format!("{dir}/domain/{name}");
805 return Ok(socket_path);
806 }
807 return Ok(rest.to_string());
808 }
809
810 // Handle named pipe format
811 if let Some(rest) = descriptor.strip_prefix("tab.pipe://") {
812 // Format: tab.pipe://<host>/pipe/<name>
813 // Return as pipe path: \\<host>\pipe\<name>
814 if let Some(idx) = rest.find("/pipe/") {
815 let host = &rest[..idx];
816 let name = &rest[idx + 6..]; // "/pipe/".len() == 6
817 let pipe_path = format!(r"\\{host}\pipe\{name}");
818 return Ok(pipe_path);
819 }
820 return Ok(rest.to_string());
821 }
822
823 // Handle TCP prefixes
824 let without_prefix = descriptor
825 .strip_prefix("tab.tcp://")
826 .or_else(|| descriptor.strip_prefix("tcp.grpc://"))
827 .or_else(|| descriptor.strip_prefix("tcp.grpctls://"))
828 .or_else(|| descriptor.strip_prefix("tcp://"))
829 .unwrap_or(descriptor);
830
831 // Validate it looks like host:port
832 if without_prefix.contains(':') && !without_prefix.is_empty() {
833 Ok(without_prefix.to_string())
834 } else {
835 Err(Error::internal(format!(
836 "Invalid connection descriptor format: '{descriptor}'. Expected '<scheme>://host:port' or 'tab.domain://<dir>/domain/<name>'"
837 )))
838 }
839 }
840
841 /// Returns the libpq endpoint for connecting to this instance.
842 ///
843 /// The endpoint is in the format "host:port" (e.g., "localhost:54321").
844 ///
845 /// Returns `None` if the process was started in gRPC-only mode.
846 /// Use [`grpc_endpoint`](Self::grpc_endpoint) for gRPC connections.
847 #[must_use]
848 pub fn endpoint(&self) -> Option<&str> {
849 self.endpoint.as_deref()
850 }
851
852 /// Returns the libpq endpoint, or an error if not available.
853 ///
854 /// This is a convenience method to avoid `unwrap()` calls when you need
855 /// the endpoint and want proper error handling.
856 ///
857 /// # Errors
858 ///
859 /// Returns an error if this process was started in gRPC-only mode.
860 ///
861 /// # Example
862 ///
863 /// ```no_run
864 /// use hyperdb_api::{HyperProcess, Result};
865 ///
866 /// fn main() -> Result<()> {
867 /// let hyper = HyperProcess::new(None, None)?;
868 /// let endpoint = hyper.require_endpoint()?; // No unwrap() needed!
869 /// println!("Server running at: {}", endpoint);
870 /// Ok(())
871 /// }
872 /// ```
873 pub fn require_endpoint(&self) -> crate::error::Result<&str> {
874 self.endpoint().ok_or_else(|| {
875 crate::error::Error::internal(
876 "HyperProcess does not have a libpq endpoint (gRPC-only mode). \
877 Use grpc_endpoint() instead or start with LibPq or Both listen mode.",
878 )
879 })
880 }
881
882 /// Returns the gRPC endpoint for connecting to this instance.
883 ///
884 /// The endpoint is in the format "host:port" (e.g., "127.0.0.1:7484").
885 ///
886 /// Returns `None` if the process was started in libpq-only mode.
887 #[must_use]
888 pub fn grpc_endpoint(&self) -> Option<&str> {
889 self.grpc_endpoint.as_deref()
890 }
891
892 /// Returns the gRPC endpoint, or an error if not available.
893 ///
894 /// This is a convenience method to avoid `unwrap()` calls when you need
895 /// the gRPC endpoint and want proper error handling.
896 ///
897 /// # Errors
898 ///
899 /// Returns an error if this process was started in libpq-only mode.
900 pub fn require_grpc_endpoint(&self) -> crate::error::Result<&str> {
901 self.grpc_endpoint().ok_or_else(|| {
902 crate::error::Error::internal(
903 "HyperProcess does not have a gRPC endpoint (libpq-only mode). \
904 Use endpoint() instead or start with Grpc or Both listen mode.",
905 )
906 })
907 }
908
909 /// Returns the gRPC endpoint as a full URL suitable for gRPC clients.
910 ///
911 /// Returns the endpoint prefixed with "http://" (e.g., "<http://127.0.0.1:7484>").
912 /// Returns `None` if the process was started in libpq-only mode.
913 #[must_use]
914 pub fn grpc_url(&self) -> Option<String> {
915 self.grpc_endpoint.as_ref().map(|ep| format!("http://{ep}"))
916 }
917
918 /// Returns the gRPC URL, or an error if not available.
919 ///
920 /// This is a convenience method to avoid `unwrap()` calls when you need
921 /// the gRPC URL and want proper error handling.
922 ///
923 /// # Errors
924 ///
925 /// Returns an error if this process was started in libpq-only mode.
926 pub fn require_grpc_url(&self) -> crate::error::Result<String> {
927 Ok(format!("http://{}", self.require_grpc_endpoint()?))
928 }
929
930 /// Returns the listen mode this process was started with.
931 #[must_use]
932 pub fn listen_mode(&self) -> ListenMode {
933 self.listen_mode
934 }
935
936 /// Returns the transport mode this process was started with.
937 #[must_use]
938 pub fn transport_mode(&self) -> TransportMode {
939 self.transport_mode
940 }
941
942 /// Returns the connection endpoint for this process.
943 ///
944 /// This returns a [`ConnectionEndpoint`] that can be used to connect
945 /// to this Hyper instance via TCP, Unix Domain Socket, or Named Pipe.
946 #[must_use]
947 pub fn connection_endpoint(&self) -> Option<&ConnectionEndpoint> {
948 self.connection_endpoint.as_ref()
949 }
950
951 /// Returns the log directory where hyperd writes its log files.
952 ///
953 /// The log file is typically `hyperd.log` within this directory.
954 /// Returns `None` if the log directory could not be determined (e.g.,
955 /// default parameters were disabled and no `log_dir` was specified).
956 ///
957 /// This is useful for setting up [`QueryStatsProvider`](crate::QueryStatsProvider)
958 /// implementations that parse the Hyper log file.
959 #[must_use]
960 pub fn log_dir(&self) -> Option<&Path> {
961 self.log_dir.as_deref()
962 }
963
964 /// Returns the socket directory used for UDS connections (Unix only).
965 ///
966 /// Returns `None` if the process is using TCP or if no socket directory was created.
967 #[cfg(unix)]
968 #[must_use]
969 pub fn socket_directory(&self) -> Option<&Path> {
970 self.socket_directory.as_deref()
971 }
972
973 /// Returns the pipe name used for Named Pipe connections (Windows only).
974 ///
975 /// Returns `None` if the process is using TCP.
976 #[cfg(windows)]
977 #[must_use]
978 pub fn pipe_name(&self) -> Option<&str> {
979 self.pipe_name.as_deref()
980 }
981
982 /// Returns the process ID of the Hyper server.
983 #[must_use]
984 pub fn pid(&self) -> Option<u32> {
985 self.child.as_ref().map(std::process::Child::id)
986 }
987
988 /// Returns whether the Hyper server process is still running.
989 #[must_use]
990 pub fn is_running(&self) -> bool {
991 if let Some(ref child) = self.child {
992 // Try to check if process is alive without waiting
993 #[cfg(unix)]
994 {
995 match Command::new("kill")
996 .args(["-0", &child.id().to_string()])
997 .output()
998 {
999 Ok(output) => output.status.success(),
1000 Err(_) => false,
1001 }
1002 }
1003 #[cfg(not(unix))]
1004 {
1005 // On Windows, we can't easily check without waiting
1006 // Assume it's running if we have a handle
1007 let _ = child; // Silence unused variable warning
1008 true
1009 }
1010 } else {
1011 false
1012 }
1013 }
1014
1015 /// Returns true if the hyperd child process has exited (or no child exists).
1016 ///
1017 /// Uses [`std::process::Child::try_wait`] under the hood, which is correct
1018 /// on both Unix and Windows. On Unix this also reaps any zombie state as a
1019 /// side effect — a hyperd that has been SIGKILLed but not yet `wait()`ed
1020 /// on by the parent will be observed as exited and cleaned up here.
1021 ///
1022 /// Prefer this over [`Self::is_running`] when the caller owns the
1023 /// `HyperProcess` mutably and needs an authoritative liveness signal.
1024 /// `is_running` uses `kill -0` on Unix (which incorrectly reports zombies
1025 /// as alive) and is a no-op on Windows.
1026 pub fn has_exited(&mut self) -> bool {
1027 match self.child.as_mut() {
1028 Some(child) => match child.try_wait() {
1029 Ok(Some(_status)) => true,
1030 Ok(None) => false,
1031 Err(_) => true,
1032 },
1033 None => true,
1034 }
1035 }
1036
1037 /// Shuts down the Hyper server gracefully with a timeout.
1038 ///
1039 /// This closes the callback connection, which signals Hyper to shut down gracefully.
1040 /// If Hyper doesn't exit within the timeout, it will be forcefully terminated.
1041 ///
1042 /// # Arguments
1043 ///
1044 /// * `timeout` - Maximum time to wait for graceful shutdown before force-killing.
1045 ///
1046 /// # Errors
1047 ///
1048 /// Returns an error if the shutdown fails.
1049 pub fn shutdown_timeout(mut self, timeout: Duration) -> Result<()> {
1050 self.shutdown_initiated.store(true, Ordering::SeqCst);
1051 self.do_shutdown(Some(timeout))
1052 }
1053
1054 /// Shuts down the Hyper server gracefully, waiting indefinitely.
1055 ///
1056 /// This closes the callback connection and waits for Hyper to exit.
1057 /// Use [`shutdown_timeout`](Self::shutdown_timeout) if you need a timeout.
1058 ///
1059 /// # Errors
1060 ///
1061 /// Returns an error if the shutdown fails.
1062 pub fn shutdown_graceful(mut self) -> Result<()> {
1063 self.shutdown_initiated.store(true, Ordering::SeqCst);
1064 self.do_shutdown(None)
1065 }
1066
1067 /// Closes the callback connection to signal Hyper to shut down.
1068 ///
1069 /// This is the graceful shutdown mechanism - Hyper monitors the callback connection
1070 /// and will initiate shutdown when it's closed.
1071 fn close_callback_connection(&mut self) {
1072 if let Some(conn) = self.callback_connection.take() {
1073 // Gracefully shutdown both directions
1074 let _ = conn.shutdown(Shutdown::Both);
1075 // Connection is dropped here, closing the socket
1076 }
1077 }
1078
1079 /// Internal shutdown implementation.
1080 fn do_shutdown(&mut self, timeout: Option<Duration>) -> Result<()> {
1081 info!(target: "hyperdb_api", "hyperd-shutdown");
1082
1083 // Step 1: Close the callback connection to signal graceful shutdown
1084 // Hyper will detect this and begin shutting down
1085 self.close_callback_connection();
1086
1087 if let Some(mut child) = self.child.take() {
1088 // Step 2: Wait for the process to exit
1089 let wait_result = if let Some(timeout) = timeout {
1090 // Wait with timeout
1091 let start = std::time::Instant::now();
1092 loop {
1093 match child.try_wait() {
1094 Ok(Some(status)) => break Ok(status),
1095 Ok(None) => {
1096 if start.elapsed() > timeout {
1097 // Step 3: Force kill ONLY after timeout
1098 // This should rarely happen if Hyper is healthy
1099 #[cfg(unix)]
1100 {
1101 // Try SIGTERM first
1102 let _ = Command::new("kill")
1103 .args(["-TERM", &child.id().to_string()])
1104 .output();
1105 thread::sleep(Duration::from_millis(100));
1106 }
1107 // Then force kill
1108 let _ = child.kill();
1109 break child.wait().map_err(|e| {
1110 Error::connection_with_io("Failed to wait for hyperd", e)
1111 });
1112 }
1113 thread::sleep(Duration::from_millis(100));
1114 }
1115 Err(e) => {
1116 break Err(Error::connection_with_io("Failed to wait for hyperd", e));
1117 }
1118 }
1119 }
1120 } else {
1121 // Wait indefinitely
1122 child
1123 .wait()
1124 .map_err(|e| Error::connection_with_io("Failed to wait for hyperd", e))
1125 };
1126
1127 wait_result?;
1128 }
1129
1130 Ok(())
1131 }
1132}
1133
1134impl Drop for HyperProcess {
1135 fn drop(&mut self) {
1136 if !self.shutdown_initiated.load(Ordering::SeqCst) {
1137 // Try to gracefully shutdown with a short timeout
1138 let _ = self.do_shutdown(Some(Duration::from_secs(5)));
1139 }
1140
1141 // Clean up socket directory if we created one
1142 #[cfg(unix)]
1143 if let Some(ref dir) = self.socket_directory {
1144 // Only clean up if it's a temp directory we created (contains our PID)
1145 let dir_name = dir.file_name().and_then(|n| n.to_str()).unwrap_or("");
1146 if dir_name.starts_with("hyper-") {
1147 let _ = std::fs::remove_dir_all(dir);
1148 }
1149 }
1150 }
1151}
1152
1153// SAFETY: `HyperProcess` owns its `std::process::Child` handle and (optionally)
1154// a TCP connection. Both are themselves `Send`, and no field holds thread-local
1155// state or a non-`Send` raw pointer. Ownership of a `HyperProcess` therefore
1156// transfers cleanly across thread boundaries.
1157unsafe impl Send for HyperProcess {}
1158
1159/// Special parameter key that disables default instance parameters when present.
1160///
1161/// By default, [`HyperProcess`] starts hyperd with a set of sensible default parameters
1162/// (matching the C++ `HyperProcess` behavior). If you need full control over all parameters,
1163/// include this key in your [`Parameters`] to disable all defaults.
1164pub(crate) const NO_DEFAULT_PARAMETERS: &str = "no_default_parameters";
1165
1166/// Default log configuration for hyperd: file-based JSON logging.
1167const DEFAULT_LOG_CONFIG: &str = "file,json,all,hyperd,0";
1168
1169/// Parameters for configuring the Hyper server.
1170///
1171/// When starting a [`HyperProcess`], a set of default parameters are automatically applied
1172/// (matching the C++ `HyperProcess` behavior):
1173///
1174/// | Parameter | Default Value | Description |
1175/// |-----------|---------------|-------------|
1176/// | `init_user` | `tableau_internal_user` | Initial user for the Hyper instance |
1177/// | `language` | `en_US` | Default language setting |
1178/// | `log_config` | `file,json,all,hyperd,0` | Log configuration |
1179/// | `date_style` | `MDY` | Date format (Month-Day-Year) |
1180/// | `date_style_lenient` | `false` | Strict date parsing |
1181/// | `log_dir` | Current directory | Log file directory |
1182/// | `no_password` | (flag) | Disable password requirement |
1183/// | `skip_license` | (flag) | Skip license check |
1184/// | `default_database_version` | `3` | File format version for newly created `.hyper` databases (v3 adds 128-bit NUMERIC support, required for DECIMAL128 parquet columns) |
1185///
1186/// To disable these defaults, add the `no_default_parameters` key (for example
1187/// `params.set("no_default_parameters", "")` via [`Parameters::set`].
1188///
1189/// # Listen Modes
1190///
1191/// Use [`set_listen_mode`](Parameters::set_listen_mode) to configure which protocols Hyper listens on:
1192///
1193/// ```
1194/// use hyperdb_api::{ListenMode, Parameters};
1195///
1196/// // gRPC only (for Arrow-based queries)
1197/// let mut params = Parameters::new();
1198/// params.set_listen_mode(ListenMode::Grpc { port: 0 });
1199///
1200/// // Both libpq and gRPC
1201/// let mut params = Parameters::new();
1202/// params.set_listen_mode(ListenMode::Both { grpc_port: 7484 });
1203/// ```
1204///
1205/// # Example
1206///
1207/// ```
1208/// use hyperdb_api::Parameters;
1209///
1210/// let mut params = Parameters::new();
1211/// params.set("log_file_size_limit", "100k");
1212/// params.set("log_file_max_count", "7");
1213/// ```
1214///
1215/// # Transport Modes
1216///
1217/// Use [`set_transport_mode`](Parameters::set_transport_mode) to control whether Hyper uses
1218/// TCP or IPC (Unix Domain Sockets):
1219///
1220/// ```
1221/// use hyperdb_api::{Parameters, TransportMode};
1222///
1223/// let mut params = Parameters::new();
1224/// params.set_transport_mode(TransportMode::Tcp); // Force TCP instead of IPC
1225/// ```
1226///
1227/// Transport mode for `HyperProcess` connections.
1228///
1229/// Controls whether the server uses TCP or Unix Domain Sockets (IPC) for connections.
1230/// On Unix systems, IPC is the default for better local performance.
1231/// On Windows, TCP is always used.
1232#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
1233pub enum TransportMode {
1234 /// Use IPC (Unix Domain Sockets on Unix, Named Pipes on Windows).
1235 /// This is the default mode and provides better performance for local connections.
1236 #[default]
1237 Ipc,
1238
1239 /// Use TCP/IP connections.
1240 /// Required when connecting from remote clients or when IPC is not available.
1241 Tcp,
1242}
1243
1244/// Parameters for configuring the Hyper server.
1245///
1246/// When starting a [`HyperProcess`], a set of default parameters are automatically applied
1247/// (matching the C++ `HyperProcess` behavior). You can override these defaults or disable
1248/// them entirely by adding the `no_default_parameters` key (for example
1249/// `params.set("no_default_parameters", "")` via [`Parameters::set`].
1250///
1251/// # Transport Modes
1252///
1253/// Use [`set_transport_mode`](Self::set_transport_mode) to control whether Hyper uses
1254/// TCP or IPC (Unix Domain Sockets on Unix systems).
1255///
1256/// # Example
1257///
1258/// ```
1259/// use hyperdb_api::{Parameters, TransportMode};
1260///
1261/// let mut params = Parameters::new();
1262/// params.set("log_file_size_limit", "100k");
1263/// params.set_transport_mode(TransportMode::Tcp); // Force TCP instead of IPC
1264/// ```
1265#[derive(Debug, Clone, Default)]
1266pub struct Parameters {
1267 values: Vec<(String, String)>,
1268 /// The listen mode for the Hyper server.
1269 pub(crate) listen_mode: Option<ListenMode>,
1270 /// The transport mode (TCP or IPC/UDS).
1271 pub(crate) transport_mode: Option<TransportMode>,
1272 /// Custom domain socket directory (Unix only).
1273 #[cfg(unix)]
1274 pub(crate) domain_socket_directory: Option<PathBuf>,
1275}
1276
1277impl Parameters {
1278 /// Creates a new empty Parameters instance.
1279 #[must_use]
1280 pub fn new() -> Self {
1281 Parameters {
1282 values: Vec::new(),
1283 listen_mode: None,
1284 transport_mode: None,
1285 #[cfg(unix)]
1286 domain_socket_directory: None,
1287 }
1288 }
1289
1290 /// Sets the transport mode (TCP or IPC/UDS).
1291 ///
1292 /// By default, `HyperProcess` uses IPC (Unix Domain Sockets on Unix) for better
1293 /// performance. Use `TransportMode::Tcp` if you need TCP connections.
1294 ///
1295 /// # Example
1296 ///
1297 /// ```
1298 /// use hyperdb_api::{Parameters, TransportMode};
1299 ///
1300 /// let mut params = Parameters::new();
1301 /// params.set_transport_mode(TransportMode::Tcp); // Use TCP instead of IPC
1302 /// ```
1303 pub fn set_transport_mode(&mut self, mode: TransportMode) -> &mut Self {
1304 self.transport_mode = Some(mode);
1305 self
1306 }
1307
1308 /// Returns the configured transport mode.
1309 #[must_use]
1310 pub fn transport_mode(&self) -> Option<TransportMode> {
1311 self.transport_mode
1312 }
1313
1314 /// Sets a custom domain socket directory (Unix only).
1315 ///
1316 /// By default, `HyperProcess` creates sockets in a temporary directory.
1317 /// Use this to specify a custom location.
1318 #[cfg(unix)]
1319 pub fn set_domain_socket_directory(&mut self, dir: impl Into<PathBuf>) -> &mut Self {
1320 self.domain_socket_directory = Some(dir.into());
1321 self
1322 }
1323
1324 /// Returns the configured domain socket directory (Unix only).
1325 #[cfg(unix)]
1326 #[must_use]
1327 pub fn domain_socket_directory(&self) -> Option<&Path> {
1328 self.domain_socket_directory.as_deref()
1329 }
1330
1331 /// Sets a parameter value.
1332 ///
1333 /// # Arguments
1334 ///
1335 /// * `key` - The parameter name.
1336 /// * `value` - The parameter value (empty string for flags).
1337 pub fn set(&mut self, key: impl Into<String>, value: impl Into<String>) -> &mut Self {
1338 let key = key.into();
1339 let value = value.into();
1340
1341 // Update existing or add new
1342 if let Some(entry) = self.values.iter_mut().find(|(k, _)| k == &key) {
1343 entry.1 = value;
1344 } else {
1345 self.values.push((key, value));
1346 }
1347
1348 self
1349 }
1350
1351 /// Sets the listen mode for the Hyper server.
1352 ///
1353 /// This controls which protocols the server listens on:
1354 /// - [`ListenMode::LibPq`]: `PostgreSQL` wire protocol only (default)
1355 /// - [`ListenMode::Grpc`]: gRPC protocol only (query-only, Arrow results)
1356 /// - [`ListenMode::Both`]: Both protocols enabled
1357 ///
1358 /// # Example
1359 ///
1360 /// ```
1361 /// use hyperdb_api::{ListenMode, Parameters};
1362 ///
1363 /// let mut params = Parameters::new();
1364 /// params.set_listen_mode(ListenMode::Grpc { port: 0 }); // Auto-assign port
1365 /// ```
1366 pub fn set_listen_mode(&mut self, mode: ListenMode) -> &mut Self {
1367 self.listen_mode = Some(mode);
1368 self
1369 }
1370
1371 /// Returns the configured listen mode, if any.
1372 #[must_use]
1373 pub fn listen_mode(&self) -> Option<ListenMode> {
1374 self.listen_mode
1375 }
1376
1377 /// Gets a parameter value.
1378 #[must_use]
1379 pub fn get(&self, key: &str) -> Option<&str> {
1380 self.values
1381 .iter()
1382 .find(|(k, _)| k == key)
1383 .map(|(_, v)| v.as_str())
1384 }
1385
1386 /// Returns whether the parameters contain the given key.
1387 #[must_use]
1388 pub fn contains_key(&self, key: &str) -> bool {
1389 self.values.iter().any(|(k, _)| k == key)
1390 }
1391
1392 /// Returns an iterator over the parameters.
1393 pub fn iter(&self) -> impl Iterator<Item = (&str, &str)> {
1394 self.values.iter().map(|(k, v)| (k.as_str(), v.as_str()))
1395 }
1396
1397 /// Returns whether the parameters are empty.
1398 #[must_use]
1399 pub fn is_empty(&self) -> bool {
1400 self.values.is_empty()
1401 }
1402
1403 /// Returns the number of parameters.
1404 #[must_use]
1405 pub fn len(&self) -> usize {
1406 self.values.len()
1407 }
1408}
1409
1410#[cfg(test)]
1411mod tests {
1412 use super::*;
1413
1414 #[test]
1415 fn test_parameters() {
1416 let mut params = Parameters::new();
1417 params.set("key1", "value1");
1418 params.set("key2", "value2");
1419
1420 assert_eq!(params.get("key1"), Some("value1"));
1421 assert_eq!(params.get("key2"), Some("value2"));
1422 assert_eq!(params.get("key3"), None);
1423 assert_eq!(params.len(), 2);
1424 }
1425
1426 #[test]
1427 fn test_parameters_update() {
1428 let mut params = Parameters::new();
1429 params.set("key", "value1");
1430 params.set("key", "value2");
1431
1432 assert_eq!(params.get("key"), Some("value2"));
1433 assert_eq!(params.len(), 1);
1434 }
1435
1436 #[test]
1437 fn test_parse_connection_descriptor() {
1438 assert_eq!(
1439 HyperProcess::parse_connection_descriptor("tab.tcp://localhost:12345").unwrap(),
1440 "localhost:12345"
1441 );
1442 assert_eq!(
1443 HyperProcess::parse_connection_descriptor("tab.tcp://127.0.0.1:7483").unwrap(),
1444 "127.0.0.1:7483"
1445 );
1446 assert_eq!(
1447 HyperProcess::parse_connection_descriptor("tcp://localhost:8080").unwrap(),
1448 "localhost:8080"
1449 );
1450 // Already in host:port format
1451 assert_eq!(
1452 HyperProcess::parse_connection_descriptor("localhost:9999").unwrap(),
1453 "localhost:9999"
1454 );
1455 }
1456
1457 #[test]
1458 fn test_parse_connection_descriptor_named_pipe() {
1459 assert_eq!(
1460 HyperProcess::parse_connection_descriptor("tab.pipe://./pipe/hyper-12345").unwrap(),
1461 r"\\.\pipe\hyper-12345"
1462 );
1463 assert_eq!(
1464 HyperProcess::parse_connection_descriptor("tab.pipe://server1/pipe/mydb").unwrap(),
1465 r"\\server1\pipe\mydb"
1466 );
1467 }
1468
1469 #[test]
1470 fn test_parse_connection_descriptor_invalid() {
1471 assert!(HyperProcess::parse_connection_descriptor("").is_err());
1472 assert!(HyperProcess::parse_connection_descriptor("invalid").is_err());
1473 }
1474
1475 #[test]
1476 fn test_parameters_contains_key() {
1477 let mut params = Parameters::new();
1478 params.set("key1", "value1");
1479
1480 assert!(params.contains_key("key1"));
1481 assert!(!params.contains_key("key2"));
1482 }
1483
1484 #[test]
1485 fn test_no_default_parameters_constant() {
1486 // Verify the constant matches what C++ uses
1487 assert_eq!(NO_DEFAULT_PARAMETERS, "no_default_parameters");
1488 }
1489
1490 #[test]
1491 fn test_parameters_with_no_defaults() {
1492 let mut params = Parameters::new();
1493 params.set(NO_DEFAULT_PARAMETERS, "");
1494 params.set("init_user", "custom_user");
1495
1496 assert!(params.contains_key(NO_DEFAULT_PARAMETERS));
1497 assert_eq!(params.get("init_user"), Some("custom_user"));
1498 }
1499}