use std::collections::VecDeque;
use std::io::{Read, Write};
use std::process::{Child, ChildStdin, Command, Stdio};
use std::sync::{
Arc, Mutex, MutexGuard, OnceLock,
atomic::{AtomicBool, AtomicUsize, Ordering},
mpsc,
};
use std::thread;
use std::time::{Duration, Instant};
use mlua::{AnyUserData, Lua, MultiValue, Table, UserData, UserDataMethods, Value as LuaValue};
use crate::runtime::encoding::{RuntimeTextEncoding, decode_runtime_text, encode_runtime_text};
#[cfg(unix)]
use libc::{ESRCH, SIGKILL};
#[cfg(windows)]
use std::mem::size_of;
#[cfg(unix)]
use std::os::unix::process::CommandExt;
#[cfg(windows)]
use std::os::windows::io::{AsRawHandle, FromRawHandle, OwnedHandle};
#[cfg(windows)]
use std::os::windows::process::CommandExt;
#[cfg(windows)]
use windows_sys::Win32::Foundation::{
DUPLICATE_SAME_ACCESS, DuplicateHandle, ERROR_ACCESS_DENIED, ERROR_NO_MORE_FILES, HANDLE,
INVALID_HANDLE_VALUE, WAIT_FAILED, WAIT_OBJECT_0, WAIT_TIMEOUT,
};
#[cfg(windows)]
use windows_sys::Win32::System::Diagnostics::ToolHelp::{
CreateToolhelp32Snapshot, TH32CS_SNAPTHREAD, THREADENTRY32, Thread32First, Thread32Next,
};
#[cfg(windows)]
use windows_sys::Win32::System::JobObjects::{
AssignProcessToJobObject, CreateJobObjectW, IsProcessInJob, JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE,
JOBOBJECT_BASIC_ACCOUNTING_INFORMATION, JOBOBJECT_EXTENDED_LIMIT_INFORMATION,
JobObjectBasicAccountingInformation, JobObjectExtendedLimitInformation,
QueryInformationJobObject, SetInformationJobObject, TerminateJobObject,
};
#[cfg(windows)]
use windows_sys::Win32::System::Threading::{
CREATE_BREAKAWAY_FROM_JOB, CREATE_NEW_PROCESS_GROUP, CREATE_SUSPENDED, GetCurrentProcess,
GetExitCodeProcess, OpenThread, ResumeThread, THREAD_SUSPEND_RESUME, WaitForSingleObject,
};
const DEFAULT_SESSION_READ_TIMEOUT_MS: u64 = 100;
const DEFAULT_SESSION_CLOSE_TIMEOUT_MS: u64 = 1_000;
const DEFAULT_SESSION_MAX_READ_BYTES: usize = 64 * 1024;
const DEFAULT_SESSION_BUFFER_LIMIT_BYTES: usize = 1024 * 1024;
const MAX_SESSION_TIMEOUT_MS: u64 = u32::MAX as u64 - 1;
const FORCED_CHILD_REAP_TIMEOUT: Duration = Duration::from_secs(5);
const FORCED_CHILD_REAP_POLL: Duration = Duration::from_millis(10);
const DETACHED_CHILD_REAPER_CAPACITY: usize = 256;
static DETACHED_CHILD_REAPER: OnceLock<Result<DetachedChildReaper, String>> = OnceLock::new();
#[cfg(windows)]
const WINDOWS_PROCESS_TREE_CONVERGENCE_TIMEOUT: Duration = Duration::from_secs(5);
#[cfg(windows)]
const WINDOWS_PROCESS_TREE_CONVERGENCE_POLL: Duration = Duration::from_millis(10);
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) struct ProcessStatusSnapshot {
pub(crate) running: bool,
pub(crate) exited: bool,
pub(crate) success: Option<bool>,
pub(crate) code: Option<i32>,
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub(crate) struct ManagedProcessOutputStats {
pub(crate) buffered_bytes: usize,
pub(crate) total_bytes: u64,
pub(crate) dropped_bytes: u64,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) struct ManagedProcessSessionStatus {
pub(crate) process: ProcessStatusSnapshot,
pub(crate) stdout: ManagedProcessOutputStats,
pub(crate) stderr: ManagedProcessOutputStats,
pub(crate) closed: bool,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) struct ManagedProcessSessionLaunchOptions {
pub(crate) stdout_encoding: RuntimeTextEncoding,
pub(crate) stderr_encoding: RuntimeTextEncoding,
pub(crate) stdin_encoding: RuntimeTextEncoding,
pub(crate) buffer_limit_bytes: usize,
}
struct ProcessSessionOpenRequest {
program: String,
args: Vec<String>,
cwd: Option<String>,
stdout_encoding: RuntimeTextEncoding,
stderr_encoding: RuntimeTextEncoding,
stdin_encoding: RuntimeTextEncoding,
buffer_limit_bytes: usize,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) struct ManagedProcessSessionReadRequest {
pub(crate) timeout_ms: u64,
pub(crate) max_bytes: usize,
pub(crate) until_text: Option<String>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) struct ManagedProcessSessionReadResult {
pub(crate) stdout: String,
pub(crate) stderr: String,
pub(crate) stdout_encoding: String,
pub(crate) stderr_encoding: String,
pub(crate) stdout_lossy: bool,
pub(crate) stderr_lossy: bool,
pub(crate) stdout_base64: Option<String>,
pub(crate) stderr_base64: Option<String>,
pub(crate) timed_out: bool,
pub(crate) stdout_stats: ManagedProcessOutputStats,
pub(crate) stderr_stats: ManagedProcessOutputStats,
}
struct ProcessSessionCloseRequest {
timeout_ms: u64,
}
#[derive(Debug, Default)]
struct ManagedProcessOutputBuffer {
bytes: VecDeque<u8>,
total_bytes: u64,
dropped_bytes: u64,
}
#[derive(Clone, Copy)]
enum ManagedProcessOutputStream {
Stdout,
Stderr,
}
pub(crate) trait ManagedProcessSessionObserver: Send + Sync + 'static {
fn stdout_readable(&self);
fn stderr_readable(&self);
fn exited(&self);
fn failed(&self);
}
struct ManagedProcessSessionState {
child: Mutex<Option<Child>>,
reaper_permit: Mutex<Option<DetachedChildReaperPermit>>,
final_reaper_keepalive: Mutex<Vec<Box<dyn FnOnce() + Send>>>,
process_tree: ProcessTreeController,
stdin: Mutex<Option<ChildStdin>>,
stdout_buffer: Arc<Mutex<ManagedProcessOutputBuffer>>,
stderr_buffer: Arc<Mutex<ManagedProcessOutputBuffer>>,
stdout_encoding: RuntimeTextEncoding,
stderr_encoding: RuntimeTextEncoding,
stdin_encoding: RuntimeTextEncoding,
stdout_reader: Mutex<Option<SessionPipeReader>>,
stderr_reader: Mutex<Option<SessionPipeReader>>,
closed: Mutex<bool>,
final_status: Mutex<Option<ProcessStatusSnapshot>>,
process_tree_terminated: Mutex<bool>,
observer: Option<Arc<dyn ManagedProcessSessionObserver>>,
observer_notifications_open: Arc<Mutex<bool>>,
}
struct SessionPipeReader {
handle: thread::JoinHandle<()>,
done_rx: mpsc::Receiver<()>,
done: Arc<AtomicBool>,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum ManagedProcessBackgroundThread {
StdoutReader,
StderrReader,
ExitWatcher,
}
type ManagedProcessBackgroundTask = Box<dyn FnOnce() + Send + 'static>;
type ManagedProcessBackgroundThreadSpawner = fn(
ManagedProcessBackgroundThread,
ManagedProcessBackgroundTask,
) -> Result<thread::JoinHandle<()>, std::io::Error>;
struct ProcessTreeController {
#[cfg(windows)]
job: WindowsProcessJob,
detached_guard: Mutex<Option<DetachedProcessTreeGuard>>,
}
struct DetachedProcessTreeGuard {
#[cfg(windows)]
job_handle: OwnedHandle,
}
struct DetachedChildReaper {
queue: Arc<DetachedChildReaperQueue>,
reserved: Arc<AtomicUsize>,
}
struct DetachedChildReaperQueue {
children: Mutex<Vec<DetachedChildReapRecord>>,
changed: std::sync::Condvar,
}
pub(crate) struct DetachedChildReaperPermit {
reaper: &'static DetachedChildReaper,
reserved: bool,
}
impl DetachedChildReaperPermit {
pub(crate) fn handoff_with_keepalive(
self,
child: Child,
keepalive: Option<Box<dyn FnOnce() + Send>>,
) {
self.handoff_with_resources(child, keepalive, None);
}
fn handoff_with_resources(
mut self,
child: Child,
keepalive: Option<Box<dyn FnOnce() + Send>>,
process_tree_guard: Option<DetachedProcessTreeGuard>,
) {
self.reaper
.queue
.children
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.push(DetachedChildReapRecord {
child,
_keepalive: keepalive.map(|action| DetachedChildKeepalive {
action: Some(action),
}),
process_tree_guard,
poll_error_reported: false,
});
self.reserved = false;
self.reaper.queue.changed.notify_one();
}
}
impl Drop for DetachedChildReaperPermit {
fn drop(&mut self) {
if self.reserved {
self.reaper.reserved.fetch_sub(1, Ordering::AcqRel);
}
}
}
struct DetachedChildReapRecord {
child: Child,
_keepalive: Option<DetachedChildKeepalive>,
process_tree_guard: Option<DetachedProcessTreeGuard>,
poll_error_reported: bool,
}
struct DetachedChildKeepalive {
action: Option<Box<dyn FnOnce() + Send>>,
}
impl Drop for DetachedChildKeepalive {
fn drop(&mut self) {
if let Some(action) = self.action.take() {
action();
}
}
}
fn combine_detached_child_keepalives(
actions: Vec<Box<dyn FnOnce() + Send>>,
) -> Option<Box<dyn FnOnce() + Send>> {
if actions.is_empty() {
None
} else {
Some(Box::new(move || {
for action in actions {
action();
}
}))
}
}
pub(crate) struct ManagedChildProcessTree {
controller: ProcessTreeController,
}
impl ManagedChildProcessTree {
pub(crate) fn spawn_with_keepalive(
command: &mut Command,
label: &str,
mut keepalive: Option<Box<dyn FnOnce() + Send>>,
) -> Result<(Child, Self, DetachedChildReaperPermit), String> {
let breakaway_requested = ProcessTreeController::prepare_command(command)?;
let reaper_permit = reserve_detached_child_reaper_slot()?;
let child = ProcessTreeController::spawn_prepared_command(command, breakaway_requested)
.map_err(|error| format!("spawn {label}: {error}"))?;
let controller = match ProcessTreeController::attach(&child) {
Ok(controller) => controller,
Err(attach_error) => {
let cleanup_error = terminate_spawned_child_after_attach_failure(
child,
reaper_permit,
keepalive.take(),
)
.err();
return Err(match cleanup_error {
Some(cleanup_error) => format!(
"attach {label} process tree: {attach_error}; spawned-process cleanup also failed: {cleanup_error}"
),
None => format!("attach {label} process tree: {attach_error}"),
});
}
};
drop(keepalive);
Ok((child, Self { controller }, reaper_permit))
}
pub(crate) fn terminate(&self, child: &Child) -> Result<(), String> {
self.controller.terminate(child)
}
pub(crate) fn detached_tree_is_empty(&self) -> Result<bool, String> {
self.controller.detached_tree_is_empty()
}
pub(crate) fn handoff_to_reaper(
&self,
permit: DetachedChildReaperPermit,
child: Child,
keepalive: Option<Box<dyn FnOnce() + Send>>,
) {
permit.handoff_with_resources(child, keepalive, self.controller.take_detached_guard());
}
pub(crate) fn clear_detached_guard(&self) {
self.controller.clear_detached_guard();
}
}
#[cfg(windows)]
struct WindowsProcessJob {
handle: OwnedHandle,
}
#[derive(Clone)]
pub(crate) struct ManagedProcessSessionCore {
state: Arc<ManagedProcessSessionState>,
}
pub(crate) type ManagedProcessSessionCleanup = Box<dyn FnOnce() + Send + Sync + 'static>;
pub(crate) type ManagedProcessSessionCleanupRetry = Box<dyn Fn() + Send + Sync + 'static>;
pub(crate) struct ManagedProcessSessionCleanupHandle {
cleanup: Mutex<Option<ManagedProcessSessionCleanup>>,
retry_teardown: Option<ManagedProcessSessionCleanupRetry>,
}
impl ManagedProcessSessionCleanupHandle {
#[cfg(test)]
pub(crate) fn new(cleanup: ManagedProcessSessionCleanup) -> Arc<Self> {
Arc::new(Self {
cleanup: Mutex::new(Some(cleanup)),
retry_teardown: None,
})
}
pub(crate) fn new_with_retry(
cleanup: ManagedProcessSessionCleanup,
retry_teardown: ManagedProcessSessionCleanupRetry,
) -> Arc<Self> {
Arc::new(Self {
cleanup: Mutex::new(Some(cleanup)),
retry_teardown: Some(retry_teardown),
})
}
pub(crate) fn run_once(&self) {
let cleanup = self
.cleanup
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.take();
if let Some(cleanup) = cleanup {
cleanup();
}
}
pub(crate) fn is_pending(&self) -> bool {
self.cleanup
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.is_some()
}
pub(crate) fn request_teardown_retry(&self) {
if self.is_pending()
&& let Some(retry_teardown) = self.retry_teardown.as_ref()
{
retry_teardown();
}
}
}
struct ManagedProcessSession {
core: ManagedProcessSessionCore,
managed_session_id: Option<u64>,
cleanup: Option<Arc<ManagedProcessSessionCleanupHandle>>,
}
fn lock_session_output_buffer(
buffer: &Arc<Mutex<ManagedProcessOutputBuffer>>,
) -> MutexGuard<'_, ManagedProcessOutputBuffer> {
buffer
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
fn lock_process_tree_terminated_flag(terminated: &Mutex<bool>) -> MutexGuard<'_, bool> {
terminated
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
fn lock_observer_notification_gate(open: &Mutex<bool>) -> MutexGuard<'_, bool> {
open.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
fn lock_session_reader_slot(
handle: &Mutex<Option<SessionPipeReader>>,
) -> MutexGuard<'_, Option<SessionPipeReader>> {
handle
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
fn lock_session_stdin_pipe(
stdin: &Mutex<Option<ChildStdin>>,
) -> MutexGuard<'_, Option<ChildStdin>> {
stdin
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
fn lock_session_closed_flag(closed: &Mutex<bool>) -> MutexGuard<'_, bool> {
closed
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
fn lock_session_final_status(
final_status: &Mutex<Option<ProcessStatusSnapshot>>,
) -> MutexGuard<'_, Option<ProcessStatusSnapshot>> {
final_status
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
fn lock_session_child(child: &Mutex<Option<Child>>) -> MutexGuard<'_, Option<Child>> {
child
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
impl ManagedProcessSessionCore {
pub(crate) fn install_final_reaper_keepalive(
&self,
keepalive: Box<dyn FnOnce() + Send>,
) -> Result<(), String> {
let mut slot = self
.state
.final_reaper_keepalive
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
slot.push(keepalive);
Ok(())
}
pub(crate) fn launch(
command: Command,
options: ManagedProcessSessionLaunchOptions,
) -> Result<Self, String> {
Self::launch_with_attacher_and_observer(
command,
options,
None,
None,
ProcessTreeController::attach,
spawn_managed_process_background_thread,
)
}
pub(crate) fn launch_with_optional_observer_and_keepalive(
command: Command,
options: ManagedProcessSessionLaunchOptions,
observer: Option<Arc<dyn ManagedProcessSessionObserver>>,
keepalive: Option<Box<dyn FnOnce() + Send>>,
) -> Result<Self, String> {
Self::launch_with_attacher_and_observer(
command,
options,
observer,
keepalive,
ProcessTreeController::attach,
spawn_managed_process_background_thread,
)
}
#[cfg(test)]
fn launch_with_observer(
command: Command,
options: ManagedProcessSessionLaunchOptions,
observer: Arc<dyn ManagedProcessSessionObserver>,
) -> Result<Self, String> {
Self::launch_with_optional_observer_and_keepalive(command, options, Some(observer), None)
}
#[cfg(test)]
fn launch_with_attacher<F>(
command: Command,
options: ManagedProcessSessionLaunchOptions,
attach: F,
) -> Result<Self, String>
where
F: FnOnce(&Child) -> Result<ProcessTreeController, String>,
{
Self::launch_with_attacher_and_observer(
command,
options,
None,
None,
attach,
spawn_managed_process_background_thread,
)
}
fn launch_with_attacher_and_observer<F>(
mut command: Command,
options: ManagedProcessSessionLaunchOptions,
observer: Option<Arc<dyn ManagedProcessSessionObserver>>,
mut final_reaper_keepalive: Option<Box<dyn FnOnce() + Send>>,
attach: F,
spawn_thread: ManagedProcessBackgroundThreadSpawner,
) -> Result<Self, String>
where
F: FnOnce(&Child) -> Result<ProcessTreeController, String>,
{
if options.buffer_limit_bytes == 0 {
return Err("buffer_limit_bytes must be greater than zero".to_string());
}
let breakaway_requested = ProcessTreeController::prepare_command(&mut command)?;
let reaper_permit = reserve_detached_child_reaper_slot()?;
command
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped());
let mut child =
ProcessTreeController::spawn_prepared_command(&mut command, breakaway_requested)
.map_err(|error| format!("spawn managed process: {error}"))?;
let process_tree = match attach(&child) {
Ok(process_tree) => process_tree,
Err(attach_error) => {
let cleanup_error = terminate_spawned_child_after_attach_failure(
child,
reaper_permit,
final_reaper_keepalive.take(),
)
.err();
return Err(match cleanup_error {
Some(cleanup_error) => format!(
"attach managed process tree: {attach_error}; spawned-process cleanup also failed: {cleanup_error}"
),
None => format!("attach managed process tree: {attach_error}"),
});
}
};
let stdin = child.stdin.take();
let stdout_pipe = child.stdout.take();
let stderr_pipe = child.stderr.take();
let stdout_buffer = Arc::new(Mutex::new(ManagedProcessOutputBuffer::default()));
let stderr_buffer = Arc::new(Mutex::new(ManagedProcessOutputBuffer::default()));
let observer_notifications_open = Arc::new(Mutex::new(observer.is_some()));
let state = Arc::new(ManagedProcessSessionState {
child: Mutex::new(Some(child)),
reaper_permit: Mutex::new(Some(reaper_permit)),
final_reaper_keepalive: Mutex::new(final_reaper_keepalive.into_iter().collect()),
process_tree,
stdin: Mutex::new(stdin),
stdout_buffer,
stderr_buffer,
stdout_encoding: options.stdout_encoding,
stderr_encoding: options.stderr_encoding,
stdin_encoding: options.stdin_encoding,
stdout_reader: Mutex::new(None),
stderr_reader: Mutex::new(None),
closed: Mutex::new(false),
final_status: Mutex::new(None),
process_tree_terminated: Mutex::new(false),
observer,
observer_notifications_open,
});
if let Some(stdout) = stdout_pipe {
let stdout_reader = match spawn_session_pipe_reader(
stdout,
state.stdout_buffer.clone(),
options.buffer_limit_bytes,
ManagedProcessOutputStream::Stdout,
state.observer.clone(),
state.observer_notifications_open.clone(),
spawn_thread,
) {
Ok(reader) => reader,
Err(error) => {
drop(stderr_pipe);
return Err(rollback_managed_process_launch(&state, error));
}
};
*lock_session_reader_slot(&state.stdout_reader) = Some(stdout_reader);
}
if let Some(stderr) = stderr_pipe {
let stderr_reader = match spawn_session_pipe_reader(
stderr,
state.stderr_buffer.clone(),
options.buffer_limit_bytes,
ManagedProcessOutputStream::Stderr,
state.observer.clone(),
state.observer_notifications_open.clone(),
spawn_thread,
) {
Ok(reader) => reader,
Err(error) => return Err(rollback_managed_process_launch(&state, error)),
};
*lock_session_reader_slot(&state.stderr_reader) = Some(stderr_reader);
}
if let Err(error) = spawn_direct_child_exit_watcher(&state, spawn_thread) {
return Err(rollback_managed_process_launch(&state, error));
}
Ok(Self { state })
}
pub(crate) fn write_text(&self, text: &str) -> Result<(), String> {
if *lock_session_closed_flag(&self.state.closed) {
return Err("session is closed".to_string());
}
let bytes = encode_runtime_text(text, self.state.stdin_encoding)?;
let mut stdin = lock_session_stdin_pipe(&self.state.stdin);
let stdin = stdin
.as_mut()
.ok_or_else(|| "stdin is closed".to_string())?;
stdin.write_all(&bytes).map_err(|error| error.to_string())?;
stdin.flush().map_err(|error| error.to_string())
}
pub(crate) fn read(
&self,
request: &ManagedProcessSessionReadRequest,
) -> Result<ManagedProcessSessionReadResult, String> {
if request.max_bytes == 0 {
return Err("max_bytes must be greater than zero".to_string());
}
let deadline = checked_timeout_deadline(request.timeout_ms, "read timeout_ms")?;
let mut timed_out = false;
loop {
if self.has_readable_output(&request.until_text)
|| self.state.output_readers_drained()?
{
break;
}
let now = Instant::now();
if now >= deadline {
timed_out = true;
break;
}
thread::sleep((deadline - now).min(Duration::from_millis(10)));
}
let (stdout_bytes, stdout_stats) =
drain_output_buffer(&self.state.stdout_buffer, request.max_bytes);
let (stderr_bytes, stderr_stats) =
drain_output_buffer(&self.state.stderr_buffer, request.max_bytes);
let stdout = decode_runtime_text(&stdout_bytes, self.state.stdout_encoding);
let stderr = decode_runtime_text(&stderr_bytes, self.state.stderr_encoding);
Ok(ManagedProcessSessionReadResult {
stdout: stdout.text,
stderr: stderr.text,
stdout_encoding: stdout.encoding,
stderr_encoding: stderr.encoding,
stdout_lossy: stdout.lossy,
stderr_lossy: stderr.lossy,
stdout_base64: stdout.base64,
stderr_base64: stderr.base64,
timed_out,
stdout_stats,
stderr_stats,
})
}
pub(crate) fn status(&self) -> Result<ManagedProcessSessionStatus, String> {
Ok(ManagedProcessSessionStatus {
process: self.state.peek_status_snapshot()?,
stdout: output_buffer_stats(&self.state.stdout_buffer),
stderr: output_buffer_stats(&self.state.stderr_buffer),
closed: *lock_session_closed_flag(&self.state.closed),
})
}
pub(crate) fn close(&self, timeout_ms: u64) -> Result<ProcessStatusSnapshot, String> {
let deadline = checked_timeout_deadline(timeout_ms, "close timeout_ms")?;
self.state.mark_closed()?;
self.state.close_stdin_pipe()?;
loop {
if self.state.peek_status_snapshot()?.exited {
break;
}
let now = Instant::now();
if now >= deadline {
break;
}
thread::sleep((deadline - now).min(Duration::from_millis(10)));
}
let final_status = self.state.kill_process_tree_and_wait()?;
self.state.join_reader_threads()?;
Ok(final_status)
}
pub(crate) fn kill(&self) -> Result<ProcessStatusSnapshot, String> {
self.state.mark_closed()?;
self.state.close_stdin_pipe()?;
let final_status = self.state.kill_process_tree_and_wait()?;
self.state.join_reader_threads()?;
Ok(final_status)
}
fn teardown_before_userdata_cleanup(&self) -> Result<(), String> {
self.kill().map(|_| ())
}
fn has_readable_output(&self, until_text: &Option<String>) -> bool {
let stdout = lock_session_output_buffer(&self.state.stdout_buffer);
let stderr = lock_session_output_buffer(&self.state.stderr_buffer);
if stdout.bytes.is_empty() && stderr.bytes.is_empty() {
return false;
}
if let Some(marker) = until_text {
let stdout_bytes = stdout.bytes.iter().copied().collect::<Vec<_>>();
let stderr_bytes = stderr.bytes.iter().copied().collect::<Vec<_>>();
let stdout_text = decode_runtime_text(&stdout_bytes, self.state.stdout_encoding).text;
let stderr_text = decode_runtime_text(&stderr_bytes, self.state.stderr_encoding).text;
return stdout_text.contains(marker) || stderr_text.contains(marker);
}
true
}
}
fn rollback_managed_process_launch(
state: &Arc<ManagedProcessSessionState>,
launch_error: String,
) -> String {
notify_managed_process_observer(
state.observer.as_ref(),
&state.observer_notifications_open,
ManagedProcessSessionObserver::failed,
);
let rollback_core = ManagedProcessSessionCore {
state: state.clone(),
};
match rollback_core.teardown_before_userdata_cleanup() {
Ok(()) => launch_error,
Err(cleanup_error) => {
format!("{launch_error}; managed process launch rollback also failed: {cleanup_error}")
}
}
}
impl ManagedProcessSession {
fn launch_core(request: ProcessSessionOpenRequest) -> mlua::Result<ManagedProcessSessionCore> {
let mut command = Command::new(&request.program);
command.args(&request.args);
if let Some(cwd) = request.cwd.as_deref() {
command.current_dir(cwd);
}
let options = ManagedProcessSessionLaunchOptions {
stdout_encoding: request.stdout_encoding,
stderr_encoding: request.stderr_encoding,
stdin_encoding: request.stdin_encoding,
buffer_limit_bytes: request.buffer_limit_bytes,
};
let core = ManagedProcessSessionCore::launch(command, options)
.map_err(|error| mlua::Error::runtime(format!("process.session.open: {error}")))?;
Ok(core)
}
#[cfg(test)]
fn open(request: ProcessSessionOpenRequest) -> mlua::Result<Self> {
let core = Self::launch_core(request)?;
Ok(Self::from_core(core, None, None))
}
fn from_core(
core: ManagedProcessSessionCore,
cleanup: Option<Arc<ManagedProcessSessionCleanupHandle>>,
managed_session_id: Option<u64>,
) -> Self {
Self {
core,
managed_session_id,
cleanup,
}
}
fn run_cleanup_once(&self) {
if let Some(cleanup) = self.cleanup.as_ref() {
cleanup.run_once();
}
}
fn write_values(&self, values: MultiValue) -> mlua::Result<bool> {
for value in values {
let text = lua_value_to_session_text(value, "process.session.write")?;
self.core
.write_text(&text)
.map_err(|error| mlua::Error::runtime(format!("process.session.write: {error}")))?;
}
Ok(true)
}
fn read(&self, lua: &Lua, args: MultiValue) -> mlua::Result<Table> {
let request = parse_session_read_request(args)?;
let read = self
.core
.read(&request)
.map_err(|error| mlua::Error::runtime(format!("process.session.read: {error}")))?;
process_session_read_result_to_lua_table(lua, read)
}
fn status(&self, lua: &Lua) -> mlua::Result<Table> {
let status = self
.core
.status()
.map_err(|error| mlua::Error::runtime(format!("process.session.status: {error}")))?;
let table = process_session_status_to_lua_table(lua, &status)?;
if let Some(managed_session_id) = self.managed_session_id {
table.set("managed_session_id", managed_session_id)?;
}
Ok(table)
}
fn close(&self, lua: &Lua, args: MultiValue) -> mlua::Result<Table> {
let request = parse_session_close_request(args)?;
let final_status = self
.core
.close(request.timeout_ms)
.map_err(|error| mlua::Error::runtime(format!("process.session.close: {error}")))?;
self.run_cleanup_once();
process_status_snapshot_to_lua_table(lua, &final_status)
}
fn kill(&self) -> mlua::Result<bool> {
let _ = self
.core
.kill()
.map_err(|error| mlua::Error::runtime(format!("process.session.kill: {error}")))?;
self.run_cleanup_once();
Ok(true)
}
#[cfg(test)]
fn kill_child(&self) -> mlua::Result<ProcessStatusSnapshot> {
self.core
.state
.kill_process_tree_and_wait()
.map_err(|error| mlua::Error::runtime(format!("process.session.kill: {error}")))
}
#[cfg(test)]
fn close_stdin(&self, operation_name: &str) -> mlua::Result<()> {
self.core
.state
.close_stdin_pipe()
.map_err(|error| mlua::Error::runtime(format!("{operation_name}: {error}")))
}
#[cfg(test)]
fn mark_closed(&self, operation_name: &str) -> mlua::Result<()> {
self.core
.state
.mark_closed()
.map_err(|error| mlua::Error::runtime(format!("{operation_name}: {error}")))
}
#[cfg(test)]
fn join_reader_threads(&self, operation_name: &str) -> mlua::Result<()> {
self.core
.state
.join_reader_threads()
.map_err(|error| mlua::Error::runtime(format!("{operation_name}: {error}")))
}
}
impl Drop for ManagedProcessSession {
fn drop(&mut self) {
match self.core.teardown_before_userdata_cleanup() {
Ok(()) => self.run_cleanup_once(),
Err(error) => {
if let Some(cleanup) = self.cleanup.as_ref() {
cleanup.request_teardown_retry();
}
crate::runtime_logging::warn(format!(
"[LuaSkill:warn] managed process userdata teardown failed and was scheduled for retry when service-owned: {error}"
));
}
}
}
}
impl ManagedProcessSessionState {
fn observer_notifications_are_open(&self) -> bool {
*lock_observer_notification_gate(&self.observer_notifications_open)
}
fn close_observer_notifications(&self) {
let mut open = lock_observer_notification_gate(&self.observer_notifications_open);
*open = false;
}
fn cached_final_status(&self) -> Result<Option<ProcessStatusSnapshot>, String> {
let final_status = lock_session_final_status(&self.final_status);
Ok(*final_status)
}
fn store_final_status(&self, status: ProcessStatusSnapshot) -> Result<(), String> {
let mut final_status = lock_session_final_status(&self.final_status);
*final_status = Some(status);
Ok(())
}
fn reader_completed(handle: &Mutex<Option<SessionPipeReader>>) -> bool {
let reader_slot = lock_session_reader_slot(handle);
reader_slot
.as_ref()
.map(|reader| reader.done.load(Ordering::Acquire))
.unwrap_or(true)
}
fn output_readers_drained(&self) -> Result<bool, String> {
Ok(Self::reader_completed(&self.stdout_reader)
&& Self::reader_completed(&self.stderr_reader))
}
fn close_stdin_pipe(&self) -> Result<(), String> {
let mut stdin = lock_session_stdin_pipe(&self.stdin);
stdin.take();
Ok(())
}
fn mark_closed(&self) -> Result<(), String> {
let mut closed = lock_session_closed_flag(&self.closed);
*closed = true;
drop(closed);
self.close_observer_notifications();
Ok(())
}
fn terminate_process_tree_once(&self, child: &Child) -> Result<(), String> {
let mut terminated = lock_process_tree_terminated_flag(&self.process_tree_terminated);
if *terminated {
return Ok(());
}
self.process_tree.terminate(child)?;
*terminated = true;
Ok(())
}
fn peek_status_snapshot(&self) -> Result<ProcessStatusSnapshot, String> {
if let Some(status) = self.cached_final_status()? {
return Ok(status);
}
#[cfg(unix)]
{
let child = lock_session_child(&self.child);
let child = child.as_ref().ok_or_else(|| {
"managed process direct child ownership was transferred".to_string()
})?;
let mut info = std::mem::MaybeUninit::<libc::siginfo_t>::zeroed();
let result = unsafe {
libc::waitid(
libc::P_PID,
child.id() as libc::id_t,
info.as_mut_ptr(),
libc::WEXITED | libc::WNOHANG | libc::WNOWAIT,
)
};
if result != 0 {
let error = std::io::Error::last_os_error();
if error.raw_os_error() == Some(libc::ECHILD) {
return Ok(ProcessStatusSnapshot {
running: false,
exited: true,
success: None,
code: None,
});
}
return Err(format!("waitid: {error}"));
}
let info = unsafe { info.assume_init() };
let reported_pid = unsafe { info.si_pid() };
if reported_pid == 0 {
return Ok(ProcessStatusSnapshot {
running: true,
exited: false,
success: None,
code: None,
});
}
let status_code = unsafe { info.si_status() };
let signal_code = info.si_code;
let (success, code) = if signal_code == libc::CLD_EXITED {
(Some(status_code == 0), Some(status_code))
} else {
(Some(false), None)
};
Ok(ProcessStatusSnapshot {
running: false,
exited: true,
success,
code,
})
}
#[cfg(windows)]
{
let child = lock_session_child(&self.child);
let child = child.as_ref().ok_or_else(|| {
"managed process direct child ownership was transferred".to_string()
})?;
peek_windows_process_status(child.as_raw_handle() as HANDLE)
}
#[cfg(all(not(unix), not(windows)))]
{
let mut child = lock_session_child(&self.child);
let child = child.as_mut().ok_or_else(|| {
"managed process direct child ownership was transferred".to_string()
})?;
match child.try_wait().map_err(|error| error.to_string())? {
Some(status) => Ok(ProcessStatusSnapshot {
running: false,
exited: true,
success: Some(status.success()),
code: status.code(),
}),
None => Ok(ProcessStatusSnapshot {
running: true,
exited: false,
success: None,
code: None,
}),
}
}
}
fn kill_process_tree_and_wait(&self) -> Result<ProcessStatusSnapshot, String> {
let mut child_guard = lock_session_child(&self.child);
let child = child_guard
.as_mut()
.ok_or_else(|| "managed process direct child ownership was transferred".to_string())?;
self.terminate_process_tree_once(child)?;
if let Some(status) = self.cached_final_status()? {
return Ok(status);
}
let status = Some(wait_for_child_exit_until(
child,
Instant::now() + FORCED_CHILD_REAP_TIMEOUT,
"managed process direct child",
)?);
let snapshot = process_status_snapshot_from_exit_status(status);
self.store_final_status(snapshot)?;
drop(child_guard);
drop(
self.reaper_permit
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.take(),
);
drop(std::mem::take(
&mut *self
.final_reaper_keepalive
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner),
));
self.process_tree.clear_detached_guard();
Ok(snapshot)
}
fn join_one_reader(
handle: &Mutex<Option<SessionPipeReader>>,
stream_name: &'static str,
) -> Result<(), String> {
let current_thread_id = thread::current().id();
let should_take = {
let mut reader_slot = lock_session_reader_slot(handle);
if reader_slot
.as_ref()
.is_some_and(|reader| reader.handle.thread().id() == current_thread_id)
{
reader_slot.take();
return Ok(());
}
let Some(reader) = reader_slot.as_mut() else {
return Ok(());
};
match reader
.done_rx
.recv_timeout(Duration::from_millis(DEFAULT_SESSION_CLOSE_TIMEOUT_MS))
{
Ok(()) | Err(mpsc::RecvTimeoutError::Disconnected) => true,
Err(mpsc::RecvTimeoutError::Timeout) => {
return Err(format!(
"{stream_name} reader shutdown timed out after {DEFAULT_SESSION_CLOSE_TIMEOUT_MS}ms"
));
}
}
};
if should_take {
let reader = lock_session_reader_slot(handle).take();
if let Some(reader) = reader {
reader
.handle
.join()
.map_err(|_| format!("{stream_name} reader panicked"))?;
}
}
Ok(())
}
fn join_reader_threads(&self) -> Result<(), String> {
Self::join_one_reader(&self.stdout_reader, "stdout")?;
Self::join_one_reader(&self.stderr_reader, "stderr")?;
Ok(())
}
fn cleanup_on_drop(&mut self) {
drop(self.mark_closed());
drop(self.close_stdin_pipe());
let kill_result = self.kill_process_tree_and_wait();
if kill_result.is_err() {
let child = self
.child
.get_mut()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.take();
let permit = self
.reaper_permit
.get_mut()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.take();
if let (Some(mut child), Some(permit)) = (child, permit) {
let _ = child.kill();
let keepalives = std::mem::take(
self.final_reaper_keepalive
.get_mut()
.unwrap_or_else(std::sync::PoisonError::into_inner),
);
let keepalive = combine_detached_child_keepalives(keepalives);
let process_tree_guard = self.process_tree.take_detached_guard();
permit.handoff_with_resources(child, keepalive, process_tree_guard);
}
}
drop(self.join_reader_threads());
}
}
impl Drop for ManagedProcessSessionState {
fn drop(&mut self) {
self.cleanup_on_drop();
}
}
impl UserData for ManagedProcessSession {
fn add_methods<M: UserDataMethods<Self>>(methods: &mut M) {
methods.add_method("write", |_, session, values: MultiValue| {
session.write_values(values)
});
methods.add_method("read", |lua, session, args: MultiValue| {
session.read(lua, args)
});
methods.add_method("status", |lua, session, ()| session.status(lua));
methods.add_method("close", |lua, session, args: MultiValue| {
session.close(lua, args)
});
methods.add_method("kill", |_, session, ()| session.kill());
}
}
pub(crate) fn create_managed_process_session_userdata(
lua: &Lua,
core: ManagedProcessSessionCore,
cleanup: Option<Arc<ManagedProcessSessionCleanupHandle>>,
managed_session_id: Option<u64>,
) -> mlua::Result<AnyUserData> {
lua.create_userdata(ManagedProcessSession::from_core(
core,
cleanup,
managed_session_id,
))
}
pub(crate) fn create_process_session_table(
lua: &Lua,
default_encoding: RuntimeTextEncoding,
) -> mlua::Result<Table> {
let table = lua.create_table()?;
table.set(
"open",
lua.create_function(move |lua, spec: LuaValue| {
let request = parse_session_open_request(spec, default_encoding)?;
let core = ManagedProcessSession::launch_core(request)?;
create_managed_process_session_userdata(lua, core, None, None)
})?,
)?;
Ok(table)
}
fn parse_session_open_request(
value: LuaValue,
default_encoding: RuntimeTextEncoding,
) -> mlua::Result<ProcessSessionOpenRequest> {
let table = match value {
LuaValue::Table(table) => table,
other => {
return Err(mlua::Error::runtime(format!(
"process.session.open: spec must be a table, got {}",
lua_value_type_name(&other)
)));
}
};
let program = require_string_field(&table, "program", "process.session.open")?;
let args = parse_string_array_field(&table, "args", "process.session.open")?;
let cwd = parse_optional_string_field(&table, "cwd", "process.session.open")?;
let encoding = parse_optional_encoding_field(&table, "encoding", "process.session.open")?
.unwrap_or(default_encoding);
let stdout_encoding =
parse_optional_encoding_field(&table, "stdout_encoding", "process.session.open")?
.unwrap_or(encoding);
let stderr_encoding =
parse_optional_encoding_field(&table, "stderr_encoding", "process.session.open")?
.unwrap_or(encoding);
let stdin_encoding =
parse_optional_encoding_field(&table, "stdin_encoding", "process.session.open")?
.unwrap_or(encoding);
let buffer_limit_bytes = parse_optional_usize_field(
&table,
"buffer_limit_bytes",
"process.session.open",
DEFAULT_SESSION_BUFFER_LIMIT_BYTES,
)?;
Ok(ProcessSessionOpenRequest {
program,
args,
cwd,
stdout_encoding,
stderr_encoding,
stdin_encoding,
buffer_limit_bytes,
})
}
fn parse_session_read_request(args: MultiValue) -> mlua::Result<ManagedProcessSessionReadRequest> {
let mut values = args.into_iter();
let value = values.next().unwrap_or(LuaValue::Nil);
match value {
LuaValue::Nil => Ok(ManagedProcessSessionReadRequest {
timeout_ms: DEFAULT_SESSION_READ_TIMEOUT_MS,
max_bytes: DEFAULT_SESSION_MAX_READ_BYTES,
until_text: None,
}),
LuaValue::Table(table) => Ok(ManagedProcessSessionReadRequest {
timeout_ms: parse_optional_timeout_ms_field(
&table,
"timeout_ms",
"process.session.read",
DEFAULT_SESSION_READ_TIMEOUT_MS,
)?,
max_bytes: parse_optional_usize_field(
&table,
"max_bytes",
"process.session.read",
DEFAULT_SESSION_MAX_READ_BYTES,
)?,
until_text: parse_optional_string_field(&table, "until_text", "process.session.read")?,
}),
other => Err(mlua::Error::runtime(format!(
"process.session.read: options must be a table, got {}",
lua_value_type_name(&other)
))),
}
}
fn parse_session_close_request(args: MultiValue) -> mlua::Result<ProcessSessionCloseRequest> {
let mut values = args.into_iter();
let value = values.next().unwrap_or(LuaValue::Nil);
match value {
LuaValue::Nil => Ok(ProcessSessionCloseRequest {
timeout_ms: DEFAULT_SESSION_CLOSE_TIMEOUT_MS,
}),
LuaValue::Table(table) => Ok(ProcessSessionCloseRequest {
timeout_ms: parse_optional_timeout_ms_field(
&table,
"timeout_ms",
"process.session.close",
DEFAULT_SESSION_CLOSE_TIMEOUT_MS,
)?,
}),
other => Err(mlua::Error::runtime(format!(
"process.session.close: options must be a table, got {}",
lua_value_type_name(&other)
))),
}
}
impl ManagedProcessBackgroundThread {
fn name(self) -> &'static str {
match self {
Self::StdoutReader => "managed-process-stdout-reader",
Self::StderrReader => "managed-process-stderr-reader",
Self::ExitWatcher => "managed-process-exit-watcher",
}
}
}
fn spawn_managed_process_background_thread(
role: ManagedProcessBackgroundThread,
task: ManagedProcessBackgroundTask,
) -> Result<thread::JoinHandle<()>, std::io::Error> {
thread::Builder::new()
.name(role.name().to_string())
.spawn(task)
}
fn spawn_session_pipe_reader<R>(
mut reader: R,
target: Arc<Mutex<ManagedProcessOutputBuffer>>,
limit_bytes: usize,
stream: ManagedProcessOutputStream,
observer: Option<Arc<dyn ManagedProcessSessionObserver>>,
observer_notifications_open: Arc<Mutex<bool>>,
spawn_thread: ManagedProcessBackgroundThreadSpawner,
) -> Result<SessionPipeReader, String>
where
R: Read + Send + 'static,
{
let (done_tx, done_rx) = mpsc::channel();
let done = Arc::new(AtomicBool::new(false));
let done_flag = done.clone();
let role = match stream {
ManagedProcessOutputStream::Stdout => ManagedProcessBackgroundThread::StdoutReader,
ManagedProcessOutputStream::Stderr => ManagedProcessBackgroundThread::StderrReader,
};
let task: ManagedProcessBackgroundTask = Box::new(move || {
let mut chunk = [0_u8; 4096];
loop {
match reader.read(&mut chunk) {
Ok(0) => break,
Ok(count) => {
{
let mut buffer = lock_session_output_buffer(&target);
append_bounded(&mut buffer, &chunk[..count], limit_bytes);
}
notify_managed_process_observer(
observer.as_ref(),
&observer_notifications_open,
|observer| stream.notify_readable(observer),
);
}
Err(error) => {
notify_managed_process_observer(
observer.as_ref(),
&observer_notifications_open,
ManagedProcessSessionObserver::failed,
);
crate::runtime_logging::warn(format!(
"[LuaSkill:warn] process.session {} reader failed: {error}",
stream.name()
));
break;
}
}
}
done_flag.store(true, Ordering::Release);
let _ = done_tx.send(());
});
let handle = spawn_thread(role, task)
.map_err(|error| format!("spawn managed process {} reader: {error}", stream.name()))?;
Ok(SessionPipeReader {
handle,
done_rx,
done,
})
}
impl ManagedProcessOutputStream {
fn name(self) -> &'static str {
match self {
Self::Stdout => "stdout",
Self::Stderr => "stderr",
}
}
fn notify_readable(self, observer: &dyn ManagedProcessSessionObserver) {
match self {
Self::Stdout => observer.stdout_readable(),
Self::Stderr => observer.stderr_readable(),
}
}
}
fn notify_managed_process_observer<F>(
observer: Option<&Arc<dyn ManagedProcessSessionObserver>>,
observer_notifications_open: &Mutex<bool>,
notify: F,
) where
F: FnOnce(&dyn ManagedProcessSessionObserver),
{
let Some(observer) = observer else {
return;
};
let should_notify = *lock_observer_notification_gate(observer_notifications_open);
if should_notify {
notify(observer.as_ref());
}
}
fn spawn_direct_child_exit_watcher(
state: &Arc<ManagedProcessSessionState>,
spawn_thread: ManagedProcessBackgroundThreadSpawner,
) -> Result<(), String> {
if state.observer.is_none() {
return Ok(());
}
let weak_state = Arc::downgrade(state);
let task: ManagedProcessBackgroundTask = Box::new(move || {
loop {
let Some(state) = weak_state.upgrade() else {
return;
};
if !state.observer_notifications_are_open() {
return;
}
let status = match state.peek_status_snapshot() {
Ok(status) => status,
Err(error) => {
notify_managed_process_observer(
state.observer.as_ref(),
&state.observer_notifications_open,
ManagedProcessSessionObserver::failed,
);
crate::runtime_logging::warn(format!(
"[LuaSkill:warn] process.session exit watcher failed: {error}"
));
return;
}
};
if status.exited {
notify_managed_process_observer(
state.observer.as_ref(),
&state.observer_notifications_open,
ManagedProcessSessionObserver::exited,
);
return;
}
drop(state);
thread::sleep(Duration::from_millis(10));
}
});
let spawn_result = spawn_thread(ManagedProcessBackgroundThread::ExitWatcher, task);
match spawn_result {
Ok(handle) => {
drop(handle);
Ok(())
}
Err(error) => Err(format!("spawn managed process exit watcher: {error}")),
}
}
impl ProcessTreeController {
fn prepare_command(command: &mut Command) -> Result<bool, String> {
#[cfg(unix)]
{
command.process_group(0);
Ok(false)
}
#[cfg(windows)]
{
let in_job = current_process_is_in_job()?;
let creation_flags = if in_job {
CREATE_NEW_PROCESS_GROUP | CREATE_BREAKAWAY_FROM_JOB | CREATE_SUSPENDED
} else {
CREATE_NEW_PROCESS_GROUP | CREATE_SUSPENDED
};
command.creation_flags(creation_flags);
Ok(in_job)
}
#[cfg(not(any(unix, windows)))]
{
let _ = command;
Ok(false)
}
}
fn spawn_prepared_command(
command: &mut Command,
breakaway_requested: bool,
) -> Result<Child, std::io::Error> {
#[cfg(windows)]
{
match command.spawn() {
Ok(child) => Ok(child),
Err(error)
if breakaway_requested
&& error.raw_os_error() == Some(ERROR_ACCESS_DENIED as i32) =>
{
crate::runtime_logging::warn(
"[LuaSkill:warn] process.session falling back to inherited host job because CREATE_BREAKAWAY_FROM_JOB was denied"
.to_string(),
);
command.creation_flags(CREATE_NEW_PROCESS_GROUP | CREATE_SUSPENDED);
command.spawn()
}
Err(error) => Err(error),
}
}
#[cfg(not(windows))]
{
let _ = breakaway_requested;
command.spawn()
}
}
fn attach(child: &Child) -> Result<Self, String> {
#[cfg(windows)]
{
let job = WindowsProcessJob::create()?;
job.assign(child)?;
job.require_contains(child.as_raw_handle() as HANDLE, child.id())?;
let detached_guard = job.create_detached_guard()?;
resume_suspended_windows_process(child.id())?;
Ok(Self {
job,
detached_guard: Mutex::new(Some(detached_guard)),
})
}
#[cfg(not(windows))]
{
let _ = child;
Ok(Self {
detached_guard: Mutex::new(None),
})
}
}
fn terminate(&self, _child: &Child) -> Result<(), String> {
#[cfg(unix)]
{
let result = unsafe { libc::kill(-(_child.id() as i32), SIGKILL) };
if result == 0 {
return Ok(());
}
let error = std::io::Error::last_os_error();
if error.raw_os_error() == Some(ESRCH) {
return Ok(());
}
#[cfg(target_os = "macos")]
if error.raw_os_error() == Some(libc::EPERM) && macos_direct_child_has_exited(_child)? {
return Ok(());
}
Err(format!("kill process group: {error}"))
}
#[cfg(windows)]
{
self.job.terminate_and_wait_empty()
}
#[cfg(not(any(unix, windows)))]
{
let _ = _child;
Ok(())
}
}
fn take_detached_guard(&self) -> Option<DetachedProcessTreeGuard> {
self.detached_guard
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.take()
}
fn detached_tree_is_empty(&self) -> Result<bool, String> {
let guard = self
.detached_guard
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
match guard.as_ref() {
Some(guard) => guard.is_empty(),
None => Ok(true),
}
}
fn clear_detached_guard(&self) {
let _released_guard = self.take_detached_guard();
}
}
#[cfg(target_os = "macos")]
fn macos_direct_child_has_exited(child: &Child) -> Result<bool, String> {
let mut info = std::mem::MaybeUninit::<libc::siginfo_t>::zeroed();
let result = unsafe {
libc::waitid(
libc::P_PID,
child.id() as libc::id_t,
info.as_mut_ptr(),
libc::WEXITED | libc::WNOHANG | libc::WNOWAIT,
)
};
if result != 0 {
let error = std::io::Error::last_os_error();
if error.raw_os_error() == Some(libc::ECHILD) {
return Ok(true);
}
return Err(format!("waitid after macOS process-group EPERM: {error}"));
}
let info = unsafe { info.assume_init() };
Ok(unsafe { info.si_pid() } != 0)
}
fn terminate_spawned_child_after_attach_failure(
mut child: Child,
reaper_permit: DetachedChildReaperPermit,
keepalive: Option<Box<dyn FnOnce() + Send>>,
) -> Result<(), String> {
#[cfg(unix)]
let tree_result = {
let result = unsafe { libc::kill(-(child.id() as i32), SIGKILL) };
if result == 0 {
Ok(())
} else {
let error = std::io::Error::last_os_error();
if error.raw_os_error() == Some(ESRCH) {
Ok(())
} else {
Err(format!("kill unattached process group: {error}"))
}
}
};
#[cfg(windows)]
let tree_result = match child.try_wait() {
Ok(Some(_)) => Ok(()),
Ok(None) => child.kill().map_err(|error| error.to_string()),
Err(probe_error) => match child.kill() {
Ok(()) => Err(format!(
"probe unattached direct child before kill: {probe_error}"
)),
Err(kill_error) => Err(format!(
"probe unattached direct child before kill: {probe_error}; kill: {kill_error}"
)),
},
};
#[cfg(not(any(unix, windows)))]
let tree_result: Result<(), String> = child.kill().map_err(|error| error.to_string());
let fallback_kill_result = if tree_result.is_err() {
match child.try_wait() {
Ok(Some(_)) => Ok(()),
Ok(None) => child.kill().map_err(|error| error.to_string()),
Err(probe_error) => match child.kill() {
Ok(()) => Err(format!(
"probe direct child before fallback kill: {probe_error}"
)),
Err(kill_error) => Err(format!(
"probe direct child before fallback kill: {probe_error}; kill: {kill_error}"
)),
},
}
} else {
Ok(())
};
let wait_result = wait_for_child_exit_until(
&mut child,
Instant::now() + FORCED_CHILD_REAP_TIMEOUT,
"unattached direct child",
)
.map(|_| ());
if wait_result.is_err() {
reaper_permit.handoff_with_keepalive(child, keepalive);
} else {
drop(reaper_permit);
drop(keepalive);
}
match (tree_result, fallback_kill_result, wait_result) {
(Ok(()), Ok(()), Ok(())) | (Ok(()), Ok(()), Err(_)) => Ok(()),
(tree_result, fallback_kill_result, wait_result) => {
let mut errors = Vec::new();
if let Err(error) = tree_result {
errors.push(error);
}
if let Err(error) = fallback_kill_result {
errors.push(format!("kill direct child: {error}"));
}
if let Err(error) = wait_result {
errors.push(format!("wait direct child: {error}"));
}
Err(errors.join("; "))
}
}
}
pub(crate) fn wait_for_child_exit_until(
child: &mut Child,
deadline: Instant,
label: &str,
) -> Result<std::process::ExitStatus, String> {
loop {
if let Some(status) = child
.try_wait()
.map_err(|error| format!("poll {label}: {error}"))?
{
return Ok(status);
}
let Some(remaining) = deadline.checked_duration_since(Instant::now()) else {
return Err(format!(
"{label} did not exit before its forced-reap deadline"
));
};
thread::sleep(remaining.min(FORCED_CHILD_REAP_POLL));
}
}
fn reserve_detached_child_reaper_slot() -> Result<DetachedChildReaperPermit, String> {
let reaper = detached_child_reaper()?;
reaper
.reserved
.fetch_update(Ordering::AcqRel, Ordering::Acquire, |reserved| {
(reserved < DETACHED_CHILD_REAPER_CAPACITY).then_some(reserved + 1)
})
.map_err(|_| {
format!("detached child reaper capacity exceeded: {DETACHED_CHILD_REAPER_CAPACITY}")
})?;
Ok(DetachedChildReaperPermit {
reaper,
reserved: true,
})
}
fn detached_child_reaper() -> Result<&'static DetachedChildReaper, String> {
match DETACHED_CHILD_REAPER.get_or_init(|| {
let queue = Arc::new(DetachedChildReaperQueue {
children: Mutex::new(Vec::new()),
changed: std::sync::Condvar::new(),
});
let reserved = Arc::new(AtomicUsize::new(0));
let worker_queue = Arc::clone(&queue);
let worker_reserved = Arc::clone(&reserved);
thread::Builder::new()
.name("luaskills-detached-child-reaper".to_string())
.spawn(move || {
loop {
let queue = Arc::clone(&worker_queue);
let reserved = Arc::clone(&worker_reserved);
if std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
run_detached_child_reaper(queue, reserved);
}))
.is_ok()
{
return;
}
crate::runtime_logging::warn(
"[LuaSkill:warn] detached child reaper recovered after an internal panic"
.to_string(),
);
}
})
.map_err(|error| format!("spawn detached child reaper: {error}"))?;
Ok(DetachedChildReaper { queue, reserved })
}) {
Ok(reaper) => Ok(reaper),
Err(error) => Err(error.clone()),
}
}
fn run_detached_child_reaper(queue: Arc<DetachedChildReaperQueue>, reserved: Arc<AtomicUsize>) {
loop {
let mut children = queue
.children
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
while children.is_empty() {
children = queue
.changed
.wait(children)
.unwrap_or_else(std::sync::PoisonError::into_inner);
}
let mut reaped_children = Vec::new();
let mut index = 0;
while index < children.len() {
let tree_empty = match children[index].process_tree_guard.as_ref() {
Some(guard) => guard.is_empty(),
None => Ok(true),
};
let tree_empty = match tree_empty {
Ok(tree_empty) => tree_empty,
Err(error) => {
if !children[index].poll_error_reported {
crate::runtime_logging::warn(format!(
"[LuaSkill:warn] detached child reaper retained a tree whose convergence could not be queried: {error}"
));
children[index].poll_error_reported = true;
}
index += 1;
continue;
}
};
match children[index].child.try_wait() {
Ok(Some(_)) if tree_empty => {
reaped_children.push(children.swap_remove(index));
}
Ok(Some(_)) => index += 1,
Ok(None) => index += 1,
Err(error) => {
if !children[index].poll_error_reported {
crate::runtime_logging::warn(format!(
"[LuaSkill:warn] detached child reaper will retain and retry an unpollable killed child: {error}"
));
children[index].poll_error_reported = true;
}
index += 1;
}
}
}
drop(children);
for mut reaped in reaped_children {
let _ = reaped.child.wait();
reserved.fetch_sub(1, Ordering::AcqRel);
}
thread::sleep(FORCED_CHILD_REAP_POLL);
}
}
#[cfg(windows)]
fn current_process_is_in_job() -> Result<bool, String> {
let mut in_job = 0;
let status = unsafe { IsProcessInJob(GetCurrentProcess(), std::ptr::null_mut(), &mut in_job) };
if status == 0 {
return Err(format!(
"IsProcessInJob: {}",
std::io::Error::last_os_error()
));
}
Ok(in_job != 0)
}
#[cfg(windows)]
fn peek_windows_process_status(handle: HANDLE) -> Result<ProcessStatusSnapshot, String> {
let wait_status = unsafe { WaitForSingleObject(handle, 0) };
match wait_status {
WAIT_TIMEOUT => Ok(ProcessStatusSnapshot {
running: true,
exited: false,
success: None,
code: None,
}),
WAIT_OBJECT_0 => {
let mut exit_code = 0_u32;
let status = unsafe { GetExitCodeProcess(handle, &mut exit_code) };
if status == 0 {
return Err(format!(
"GetExitCodeProcess: {}",
std::io::Error::last_os_error()
));
}
let code = exit_code as i32;
Ok(ProcessStatusSnapshot {
running: false,
exited: true,
success: Some(code == 0),
code: Some(code),
})
}
WAIT_FAILED => Err(format!(
"WaitForSingleObject(process status): {}",
std::io::Error::last_os_error()
)),
other => Err(format!(
"WaitForSingleObject(process status) returned unexpected status {other}"
)),
}
}
#[cfg(windows)]
fn resume_suspended_windows_process(process_id: u32) -> Result<(), String> {
let deadline = Instant::now() + WINDOWS_PROCESS_TREE_CONVERGENCE_TIMEOUT;
loop {
let snapshot = unsafe { CreateToolhelp32Snapshot(TH32CS_SNAPTHREAD, 0) };
if snapshot == INVALID_HANDLE_VALUE {
return Err(format!(
"CreateToolhelp32Snapshot(threads): {}",
std::io::Error::last_os_error()
));
}
let snapshot = unsafe { OwnedHandle::from_raw_handle(snapshot as _) };
let mut entry: THREADENTRY32 = unsafe { std::mem::zeroed() };
entry.dwSize = size_of::<THREADENTRY32>() as u32;
let mut has_entry =
unsafe { Thread32First(snapshot.as_raw_handle() as HANDLE, &mut entry) } != 0;
if !has_entry {
let error = std::io::Error::last_os_error();
if error.raw_os_error() != Some(ERROR_NO_MORE_FILES as i32) {
return Err(format!("Thread32First: {error}"));
}
}
let mut resumed_threads = 0usize;
loop {
if has_entry && entry.th32OwnerProcessID == process_id {
let raw_thread =
unsafe { OpenThread(THREAD_SUSPEND_RESUME, 0, entry.th32ThreadID) };
if raw_thread.is_null() {
return Err(format!(
"OpenThread({}, process {process_id}): {}",
entry.th32ThreadID,
std::io::Error::last_os_error()
));
}
let thread_handle = unsafe { OwnedHandle::from_raw_handle(raw_thread as _) };
let previous_count =
unsafe { ResumeThread(thread_handle.as_raw_handle() as HANDLE) };
if previous_count == u32::MAX {
return Err(format!(
"ResumeThread({}, process {process_id}): {}",
entry.th32ThreadID,
std::io::Error::last_os_error()
));
}
if previous_count != 1 {
return Err(format!(
"ResumeThread({}, process {process_id}) observed unexpected suspend count {previous_count}",
entry.th32ThreadID
));
}
resumed_threads = resumed_threads.saturating_add(1);
}
if !has_entry {
break;
}
has_entry =
unsafe { Thread32Next(snapshot.as_raw_handle() as HANDLE, &mut entry) } != 0;
if !has_entry {
let error = std::io::Error::last_os_error();
if error.raw_os_error() != Some(ERROR_NO_MORE_FILES as i32) {
return Err(format!("Thread32Next: {error}"));
}
}
}
if resumed_threads > 0 {
return Ok(());
}
if Instant::now() >= deadline {
return Err(format!(
"CREATE_SUSPENDED process {process_id} exposed no resumable thread before the {}ms deadline",
WINDOWS_PROCESS_TREE_CONVERGENCE_TIMEOUT.as_millis()
));
}
thread::sleep(WINDOWS_PROCESS_TREE_CONVERGENCE_POLL);
}
}
#[cfg(windows)]
impl WindowsProcessJob {
fn create() -> Result<Self, String> {
let raw = unsafe { CreateJobObjectW(std::ptr::null(), std::ptr::null()) };
if raw.is_null() {
return Err(format!(
"CreateJobObjectW: {}",
std::io::Error::last_os_error()
));
}
let handle = unsafe { OwnedHandle::from_raw_handle(raw as _) };
let mut info: JOBOBJECT_EXTENDED_LIMIT_INFORMATION = unsafe { std::mem::zeroed() };
info.BasicLimitInformation.LimitFlags = JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE;
let status = unsafe {
SetInformationJobObject(
handle.as_raw_handle() as _,
JobObjectExtendedLimitInformation,
&info as *const _ as *const _,
size_of::<JOBOBJECT_EXTENDED_LIMIT_INFORMATION>() as u32,
)
};
if status == 0 {
return Err(format!(
"SetInformationJobObject: {}",
std::io::Error::last_os_error()
));
}
Ok(Self { handle })
}
fn create_detached_guard(&self) -> Result<DetachedProcessTreeGuard, String> {
let process = unsafe { GetCurrentProcess() };
let mut duplicate = std::ptr::null_mut();
let status = unsafe {
DuplicateHandle(
process,
self.handle.as_raw_handle() as HANDLE,
process,
&mut duplicate,
0,
0,
DUPLICATE_SAME_ACCESS,
)
};
if status == 0 || duplicate.is_null() {
return Err(format!(
"DuplicateHandle(managed Job): {}",
std::io::Error::last_os_error()
));
}
Ok(DetachedProcessTreeGuard {
job_handle: unsafe { OwnedHandle::from_raw_handle(duplicate as _) },
})
}
fn assign(&self, child: &Child) -> Result<(), String> {
let status = unsafe {
AssignProcessToJobObject(self.handle.as_raw_handle() as _, child.as_raw_handle() as _)
};
if status == 0 {
let error = std::io::Error::last_os_error();
return Err(format!(
"AssignProcessToJobObject(process {}): {error}",
child.id()
));
}
Ok(())
}
fn require_contains(&self, process: HANDLE, process_id: u32) -> Result<(), String> {
let mut in_job = 0;
let status =
unsafe { IsProcessInJob(process, self.handle.as_raw_handle() as HANDLE, &mut in_job) };
if status == 0 {
return Err(format!(
"IsProcessInJob(process {process_id}): {}",
std::io::Error::last_os_error()
));
}
if in_job == 0 {
return Err(format!(
"process {process_id} was not attached to its managed Job Object"
));
}
Ok(())
}
fn active_processes(&self) -> Result<u32, String> {
windows_job_active_processes(self.handle.as_raw_handle() as HANDLE)
}
fn terminate_and_wait_empty(&self) -> Result<(), String> {
let status = unsafe { TerminateJobObject(self.handle.as_raw_handle() as _, 1) };
let terminate_error = (status == 0).then(std::io::Error::last_os_error);
let deadline = Instant::now() + WINDOWS_PROCESS_TREE_CONVERGENCE_TIMEOUT;
loop {
let active_processes = self.active_processes()?;
if active_processes == 0 {
return Ok(());
}
if Instant::now() >= deadline {
let primary_error = terminate_error
.as_ref()
.map(|error| format!("; TerminateJobObject failed: {error}"))
.unwrap_or_default();
return Err(format!(
"managed Job Object still has {active_processes} active processes after {}ms{primary_error}",
WINDOWS_PROCESS_TREE_CONVERGENCE_TIMEOUT.as_millis()
));
}
thread::sleep(WINDOWS_PROCESS_TREE_CONVERGENCE_POLL);
}
}
}
impl DetachedProcessTreeGuard {
fn is_empty(&self) -> Result<bool, String> {
#[cfg(windows)]
{
if unsafe { TerminateJobObject(self.job_handle.as_raw_handle() as HANDLE, 1) } == 0 {
return Err(format!(
"TerminateJobObject(detached reaper): {}",
std::io::Error::last_os_error()
));
}
windows_job_active_processes(self.job_handle.as_raw_handle() as HANDLE)
.map(|active| active == 0)
}
#[cfg(not(windows))]
{
Ok(true)
}
}
}
#[cfg(windows)]
fn windows_job_active_processes(handle: HANDLE) -> Result<u32, String> {
let mut accounting = JOBOBJECT_BASIC_ACCOUNTING_INFORMATION::default();
let status = unsafe {
QueryInformationJobObject(
handle,
JobObjectBasicAccountingInformation,
&mut accounting as *mut _ as *mut core::ffi::c_void,
size_of::<JOBOBJECT_BASIC_ACCOUNTING_INFORMATION>() as u32,
std::ptr::null_mut(),
)
};
if status == 0 {
return Err(format!(
"QueryInformationJobObject: {}",
std::io::Error::last_os_error()
));
}
Ok(accounting.ActiveProcesses)
}
fn append_bounded(buffer: &mut ManagedProcessOutputBuffer, bytes: &[u8], limit_bytes: usize) {
let effective_limit = limit_bytes.max(1);
let incoming_count = u64::try_from(bytes.len()).unwrap_or(u64::MAX);
buffer.total_bytes = buffer.total_bytes.saturating_add(incoming_count);
if bytes.len() >= effective_limit {
let dropped_count = buffer
.bytes
.len()
.saturating_add(bytes.len().saturating_sub(effective_limit));
buffer.dropped_bytes = buffer
.dropped_bytes
.saturating_add(u64::try_from(dropped_count).unwrap_or(u64::MAX));
buffer.bytes.clear();
buffer
.bytes
.extend(bytes[bytes.len() - effective_limit..].iter().copied());
return;
}
let total_len = buffer.bytes.len().saturating_add(bytes.len());
if total_len > effective_limit {
let overflow = total_len - effective_limit;
drop(buffer.bytes.drain(..overflow));
buffer.dropped_bytes = buffer
.dropped_bytes
.saturating_add(u64::try_from(overflow).unwrap_or(u64::MAX));
}
buffer.bytes.extend(bytes.iter().copied());
}
fn locked_output_buffer_stats(buffer: &ManagedProcessOutputBuffer) -> ManagedProcessOutputStats {
ManagedProcessOutputStats {
buffered_bytes: buffer.bytes.len(),
total_bytes: buffer.total_bytes,
dropped_bytes: buffer.dropped_bytes,
}
}
fn output_buffer_stats(
buffer: &Arc<Mutex<ManagedProcessOutputBuffer>>,
) -> ManagedProcessOutputStats {
let guard = lock_session_output_buffer(buffer);
locked_output_buffer_stats(&guard)
}
fn drain_output_buffer(
buffer: &Arc<Mutex<ManagedProcessOutputBuffer>>,
max_bytes: usize,
) -> (Vec<u8>, ManagedProcessOutputStats) {
let mut guard = lock_session_output_buffer(buffer);
let count = max_bytes.min(guard.bytes.len());
let drained = (0..count)
.filter_map(|_| guard.bytes.pop_front())
.collect::<Vec<_>>();
let stats = locked_output_buffer_stats(&guard);
(drained, stats)
}
fn checked_timeout_deadline(timeout_ms: u64, field_name: &str) -> Result<Instant, String> {
if timeout_ms > MAX_SESSION_TIMEOUT_MS {
return Err(format!(
"{field_name} is too large; maximum is {MAX_SESSION_TIMEOUT_MS}"
));
}
Instant::now()
.checked_add(Duration::from_millis(timeout_ms))
.ok_or_else(|| format!("{field_name} is too large"))
}
fn process_status_snapshot_from_exit_status(
status: Option<std::process::ExitStatus>,
) -> ProcessStatusSnapshot {
match status {
Some(status) => ProcessStatusSnapshot {
running: false,
exited: true,
success: Some(status.success()),
code: status.code(),
},
None => ProcessStatusSnapshot {
running: true,
exited: false,
success: None,
code: None,
},
}
}
fn process_status_snapshot_to_lua_table(
lua: &Lua,
status: &ProcessStatusSnapshot,
) -> mlua::Result<Table> {
let table = lua.create_table()?;
table.set("running", status.running)?;
table.set("exited", status.exited)?;
match status.success {
Some(success) => table.set("success", success)?,
None => table.set("success", LuaValue::Nil)?,
}
match status.code {
Some(code) => table.set("code", code)?,
None => table.set("code", LuaValue::Nil)?,
}
Ok(table)
}
fn set_output_stats_on_lua_table(
table: &Table,
stream_name: &str,
stats: ManagedProcessOutputStats,
) -> mlua::Result<()> {
table.set(
format!("{stream_name}_buffered_bytes"),
stats.buffered_bytes,
)?;
table.set(format!("{stream_name}_total_bytes"), stats.total_bytes)?;
table.set(format!("{stream_name}_dropped_bytes"), stats.dropped_bytes)?;
Ok(())
}
fn process_session_read_result_to_lua_table(
lua: &Lua,
read: ManagedProcessSessionReadResult,
) -> mlua::Result<Table> {
let result = lua.create_table()?;
result.set("stdout", read.stdout)?;
result.set("stderr", read.stderr)?;
result.set("stdout_encoding", read.stdout_encoding)?;
result.set("stderr_encoding", read.stderr_encoding)?;
result.set("stdout_lossy", read.stdout_lossy)?;
result.set("stderr_lossy", read.stderr_lossy)?;
result.set("stdout_base64", read.stdout_base64)?;
result.set("stderr_base64", read.stderr_base64)?;
result.set("timed_out", read.timed_out)?;
set_output_stats_on_lua_table(&result, "stdout", read.stdout_stats)?;
set_output_stats_on_lua_table(&result, "stderr", read.stderr_stats)?;
Ok(result)
}
fn process_session_status_to_lua_table(
lua: &Lua,
status: &ManagedProcessSessionStatus,
) -> mlua::Result<Table> {
let result = process_status_snapshot_to_lua_table(lua, &status.process)?;
result.set("closed", status.closed)?;
set_output_stats_on_lua_table(&result, "stdout", status.stdout)?;
set_output_stats_on_lua_table(&result, "stderr", status.stderr)?;
Ok(result)
}
fn require_string_field(table: &Table, key: &str, fn_name: &str) -> mlua::Result<String> {
let value: LuaValue = table.get(key)?;
require_string_value(value, fn_name, key, false)
}
fn parse_optional_string_field(
table: &Table,
key: &str,
fn_name: &str,
) -> mlua::Result<Option<String>> {
let value: LuaValue = table.get(key)?;
match value {
LuaValue::Nil => Ok(None),
value => Ok(Some(require_string_value(value, fn_name, key, false)?)),
}
}
fn parse_optional_encoding_field(
table: &Table,
key: &str,
fn_name: &str,
) -> mlua::Result<Option<RuntimeTextEncoding>> {
let value: LuaValue = table.get(key)?;
match value {
LuaValue::Nil => Ok(None),
LuaValue::String(text) => {
let label = text
.to_str()
.map_err(|_| mlua::Error::runtime(format!("{fn_name}: {key} must be UTF-8")))?;
RuntimeTextEncoding::parse(label.as_ref())
.map(Some)
.map_err(|error| mlua::Error::runtime(format!("{fn_name}: {error}")))
}
other => Err(mlua::Error::runtime(format!(
"{fn_name}: {key} must be a string, got {}",
lua_value_type_name(&other)
))),
}
}
fn parse_string_array_field(table: &Table, key: &str, fn_name: &str) -> mlua::Result<Vec<String>> {
let value: LuaValue = table.get(key)?;
match value {
LuaValue::Nil => Ok(Vec::new()),
LuaValue::Table(items) => {
let mut output = Vec::new();
for pair in items.sequence_values::<LuaValue>() {
output.push(require_string_value(pair?, fn_name, key, true)?);
}
Ok(output)
}
other => Err(mlua::Error::runtime(format!(
"{fn_name}: {key} must be an array table, got {}",
lua_value_type_name(&other)
))),
}
}
fn parse_optional_timeout_ms_field(
table: &Table,
key: &str,
fn_name: &str,
default_value: u64,
) -> mlua::Result<u64> {
let value: LuaValue = table.get(key)?;
match value {
LuaValue::Nil => Ok(default_value),
LuaValue::Integer(number) if number >= 0 => Ok(number as u64),
LuaValue::Number(number)
if number.is_finite()
&& number >= 0.0
&& number.fract() == 0.0
&& number <= u64::MAX as f64 =>
{
Ok(number as u64)
}
other => Err(mlua::Error::runtime(format!(
"{fn_name}: {key} must be a non-negative integer, got {}",
lua_value_type_name(&other)
))),
}
}
fn parse_optional_u64_field(
table: &Table,
key: &str,
fn_name: &str,
default_value: u64,
) -> mlua::Result<u64> {
let value: LuaValue = table.get(key)?;
match value {
LuaValue::Nil => Ok(default_value),
LuaValue::Integer(number) if number > 0 => Ok(number as u64),
LuaValue::Number(number) if number.is_finite() && number > 0.0 => Ok(number as u64),
other => Err(mlua::Error::runtime(format!(
"{fn_name}: {key} must be a positive number, got {}",
lua_value_type_name(&other)
))),
}
}
fn parse_optional_usize_field(
table: &Table,
key: &str,
fn_name: &str,
default_value: usize,
) -> mlua::Result<usize> {
let value = parse_optional_u64_field(table, key, fn_name, default_value as u64)?;
usize::try_from(value).map_err(|_| {
mlua::Error::runtime(format!("{fn_name}: {key} is too large for this platform"))
})
}
fn lua_value_to_session_text(value: LuaValue, fn_name: &str) -> mlua::Result<String> {
match value {
LuaValue::String(text) => Ok(text
.to_str()
.map_err(|_| mlua::Error::runtime(format!("{fn_name}: string must be valid UTF-8")))?
.to_string()),
LuaValue::Integer(number) => Ok(number.to_string()),
LuaValue::Number(number) => Ok(number.to_string()),
LuaValue::Boolean(flag) => Ok(flag.to_string()),
other => Err(mlua::Error::runtime(format!(
"{fn_name}: unsupported value {}",
lua_value_type_name(&other)
))),
}
}
fn require_string_value(
value: LuaValue,
fn_name: &str,
param_name: &str,
allow_blank: bool,
) -> mlua::Result<String> {
let text = match value {
LuaValue::String(text) => text
.to_str()
.map_err(|_| {
mlua::Error::runtime(format!("{fn_name}: {param_name} must be valid UTF-8"))
})?
.to_string(),
other => {
return Err(mlua::Error::runtime(format!(
"{fn_name}: {param_name} must be a string, got {}",
lua_value_type_name(&other)
)));
}
};
if !allow_blank && text.trim().is_empty() {
return Err(mlua::Error::runtime(format!(
"{fn_name}: {param_name} must not be empty"
)));
}
if text.contains('\0') {
return Err(mlua::Error::runtime(format!(
"{fn_name}: {param_name} must not contain NUL bytes"
)));
}
Ok(text)
}
fn lua_value_type_name(value: &LuaValue) -> &'static str {
match value {
LuaValue::Nil => "nil",
LuaValue::Boolean(_) => "boolean",
LuaValue::LightUserData(_) => "lightuserdata",
LuaValue::Integer(_) | LuaValue::Number(_) => "number",
LuaValue::String(_) => "string",
LuaValue::Table(_) => "table",
LuaValue::Function(_) => "function",
LuaValue::Thread(_) => "thread",
LuaValue::UserData(_) => "userdata",
LuaValue::Error(_) => "error",
LuaValue::Other(_) => "other",
}
}
#[cfg(test)]
mod tests;