This commit is contained in:
@@ -0,0 +1,558 @@
|
||||
//! Bounded session actor for Lumbridge-owned processes.
|
||||
//!
|
||||
//! The actor is deliberately transport-neutral. It owns one local PTY today;
|
||||
//! future local IPC and remote transports can preserve the same ordered command
|
||||
//! and event contract without exposing [`lumbridge_pty::PtySession`] to a UI.
|
||||
|
||||
#![forbid(unsafe_code)]
|
||||
|
||||
pub use lumbridge_pty::{CommandConfig, PtyOptions, TerminalSize};
|
||||
use lumbridge_pty::{ExitStatus, OutputEvent, PtyError, PtySession};
|
||||
use std::num::NonZeroUsize;
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::sync::mpsc::{self, Receiver, RecvTimeoutError, SyncSender, TryRecvError, TrySendError};
|
||||
use std::thread::{self, JoinHandle};
|
||||
use std::time::Duration;
|
||||
use thiserror::Error;
|
||||
|
||||
const DEFAULT_COMMAND_QUEUE: usize = 64;
|
||||
const DEFAULT_EVENT_QUEUE: usize = 128;
|
||||
const DEFAULT_POLL_INTERVAL: Duration = Duration::from_millis(10);
|
||||
static NEXT_SESSION_ID: AtomicU64 = AtomicU64::new(1);
|
||||
|
||||
/// Process-local identity for the first runtime slice.
|
||||
///
|
||||
/// Durable local and remote session IDs belong in the persisted runtime
|
||||
/// protocol; this value only correlates events within the current process.
|
||||
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
|
||||
pub struct RuntimeSessionId(u64);
|
||||
|
||||
impl RuntimeSessionId {
|
||||
#[must_use]
|
||||
pub const fn get(self) -> u64 {
|
||||
self.0
|
||||
}
|
||||
}
|
||||
|
||||
/// A command accepted by the single-owner runtime actor.
|
||||
#[derive(Clone, Debug, Eq, PartialEq)]
|
||||
pub enum RuntimeCommand {
|
||||
Input(Vec<u8>),
|
||||
Resize(TerminalSize),
|
||||
CloseInput,
|
||||
Shutdown,
|
||||
}
|
||||
|
||||
/// The operation associated with a redacted runtime fault.
|
||||
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
|
||||
pub enum RuntimeOperation {
|
||||
Input,
|
||||
Resize,
|
||||
Output,
|
||||
Wait,
|
||||
Shutdown,
|
||||
}
|
||||
|
||||
/// An ordered event emitted by the runtime actor.
|
||||
#[derive(Clone, Debug, Eq, PartialEq)]
|
||||
pub enum RuntimeEvent {
|
||||
Started {
|
||||
session_id: RuntimeSessionId,
|
||||
process_id: Option<u32>,
|
||||
},
|
||||
Output {
|
||||
session_id: RuntimeSessionId,
|
||||
sequence: u64,
|
||||
bytes: Vec<u8>,
|
||||
},
|
||||
InputClosed {
|
||||
session_id: RuntimeSessionId,
|
||||
},
|
||||
Exited {
|
||||
session_id: RuntimeSessionId,
|
||||
status: ExitStatus,
|
||||
},
|
||||
Fault {
|
||||
session_id: RuntimeSessionId,
|
||||
operation: RuntimeOperation,
|
||||
message: String,
|
||||
},
|
||||
}
|
||||
|
||||
/// Bounded queue and polling settings for a runtime actor.
|
||||
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
|
||||
pub struct RuntimeActorOptions {
|
||||
pub command_queue: NonZeroUsize,
|
||||
pub event_queue: NonZeroUsize,
|
||||
pub poll_interval: Duration,
|
||||
}
|
||||
|
||||
impl Default for RuntimeActorOptions {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
command_queue: NonZeroUsize::new(DEFAULT_COMMAND_QUEUE).expect("non-zero constant"),
|
||||
event_queue: NonZeroUsize::new(DEFAULT_EVENT_QUEUE).expect("non-zero constant"),
|
||||
poll_interval: DEFAULT_POLL_INTERVAL,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Error)]
|
||||
pub enum RuntimeActorError {
|
||||
#[error("failed to start the PTY session: {0}")]
|
||||
Start(#[source] PtyError),
|
||||
#[error("the runtime command queue is full")]
|
||||
CommandQueueFull,
|
||||
#[error("the runtime actor is disconnected")]
|
||||
Disconnected,
|
||||
#[error("timed out waiting for a runtime event")]
|
||||
EventTimeout,
|
||||
#[error("the runtime actor thread panicked")]
|
||||
ThreadPanicked,
|
||||
#[error("the runtime actor poll interval must be greater than zero")]
|
||||
InvalidPollInterval,
|
||||
}
|
||||
|
||||
/// UI-side owner of one runtime actor and its bounded channels.
|
||||
pub struct RuntimeActor {
|
||||
session_id: RuntimeSessionId,
|
||||
process_id: Option<u32>,
|
||||
commands: Option<SyncSender<RuntimeCommand>>,
|
||||
events: Option<Receiver<RuntimeEvent>>,
|
||||
thread: Option<JoinHandle<()>>,
|
||||
}
|
||||
|
||||
impl RuntimeActor {
|
||||
/// Starts a PTY synchronously, then transfers it to its actor thread.
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// Returns [`RuntimeActorError::Start`] if the PTY or child cannot start.
|
||||
pub fn spawn(
|
||||
command: CommandConfig,
|
||||
pty_options: PtyOptions,
|
||||
actor_options: RuntimeActorOptions,
|
||||
) -> Result<Self, RuntimeActorError> {
|
||||
if actor_options.poll_interval.is_zero() {
|
||||
return Err(RuntimeActorError::InvalidPollInterval);
|
||||
}
|
||||
let session = PtySession::spawn(command, pty_options).map_err(RuntimeActorError::Start)?;
|
||||
let process_id = session.process_id();
|
||||
let session_id = RuntimeSessionId(NEXT_SESSION_ID.fetch_add(1, Ordering::Relaxed));
|
||||
let (command_sender, command_receiver) =
|
||||
mpsc::sync_channel(actor_options.command_queue.get());
|
||||
let (event_sender, event_receiver) = mpsc::sync_channel(actor_options.event_queue.get());
|
||||
let thread = thread::Builder::new()
|
||||
.name(format!("lumbridge-runtime-{}", session_id.get()))
|
||||
.spawn(move || {
|
||||
run_actor(
|
||||
session_id,
|
||||
session,
|
||||
&command_receiver,
|
||||
&event_sender,
|
||||
actor_options.poll_interval,
|
||||
);
|
||||
})
|
||||
.map_err(|error| RuntimeActorError::Start(PtyError::Io(error)))?;
|
||||
|
||||
Ok(Self {
|
||||
session_id,
|
||||
process_id,
|
||||
commands: Some(command_sender),
|
||||
events: Some(event_receiver),
|
||||
thread: Some(thread),
|
||||
})
|
||||
}
|
||||
|
||||
#[must_use]
|
||||
pub const fn session_id(&self) -> RuntimeSessionId {
|
||||
self.session_id
|
||||
}
|
||||
|
||||
#[must_use]
|
||||
pub const fn process_id(&self) -> Option<u32> {
|
||||
self.process_id
|
||||
}
|
||||
|
||||
/// Enqueues without blocking the UI thread.
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// Reports bounded backpressure or actor disconnection without returning
|
||||
/// command contents in the error value.
|
||||
pub fn try_send(&self, command: RuntimeCommand) -> Result<(), RuntimeActorError> {
|
||||
let commands = self
|
||||
.commands
|
||||
.as_ref()
|
||||
.ok_or(RuntimeActorError::Disconnected)?;
|
||||
commands.try_send(command).map_err(|error| match error {
|
||||
TrySendError::Full(_) => RuntimeActorError::CommandQueueFull,
|
||||
TrySendError::Disconnected(_) => RuntimeActorError::Disconnected,
|
||||
})
|
||||
}
|
||||
|
||||
/// Polls the next ordered runtime event without blocking.
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// Returns [`RuntimeActorError::Disconnected`] after the actor stops and all
|
||||
/// queued events have been consumed.
|
||||
pub fn try_recv(&self) -> Result<Option<RuntimeEvent>, RuntimeActorError> {
|
||||
let events = self
|
||||
.events
|
||||
.as_ref()
|
||||
.ok_or(RuntimeActorError::Disconnected)?;
|
||||
match events.try_recv() {
|
||||
Ok(event) => Ok(Some(event)),
|
||||
Err(TryRecvError::Empty) => Ok(None),
|
||||
Err(TryRecvError::Disconnected) => Err(RuntimeActorError::Disconnected),
|
||||
}
|
||||
}
|
||||
|
||||
/// Waits for the next ordered runtime event.
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// Reports timeout or actor disconnection.
|
||||
pub fn recv_timeout(&self, timeout: Duration) -> Result<RuntimeEvent, RuntimeActorError> {
|
||||
let events = self
|
||||
.events
|
||||
.as_ref()
|
||||
.ok_or(RuntimeActorError::Disconnected)?;
|
||||
match events.recv_timeout(timeout) {
|
||||
Ok(event) => Ok(event),
|
||||
Err(RecvTimeoutError::Timeout) => Err(RuntimeActorError::EventTimeout),
|
||||
Err(RecvTimeoutError::Disconnected) => Err(RuntimeActorError::Disconnected),
|
||||
}
|
||||
}
|
||||
|
||||
/// Requests graceful actor shutdown and waits for its thread.
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// Reports queue disconnection or an actor-thread panic.
|
||||
pub fn shutdown(mut self) -> Result<(), RuntimeActorError> {
|
||||
self.try_send(RuntimeCommand::Shutdown)?;
|
||||
self.commands.take();
|
||||
self.events.take();
|
||||
join_thread(&mut self.thread)
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for RuntimeActor {
|
||||
fn drop(&mut self) {
|
||||
if let Some(commands) = &self.commands {
|
||||
let _ = commands.try_send(RuntimeCommand::Shutdown);
|
||||
}
|
||||
// Disconnect both channels before joining. This releases an actor that
|
||||
// is applying event backpressure and lets its PtySession clean up.
|
||||
self.commands.take();
|
||||
self.events.take();
|
||||
let _ = join_thread(&mut self.thread);
|
||||
}
|
||||
}
|
||||
|
||||
fn join_thread(thread: &mut Option<JoinHandle<()>>) -> Result<(), RuntimeActorError> {
|
||||
let Some(thread) = thread.take() else {
|
||||
return Ok(());
|
||||
};
|
||||
thread.join().map_err(|_| RuntimeActorError::ThreadPanicked)
|
||||
}
|
||||
|
||||
fn run_actor(
|
||||
session_id: RuntimeSessionId,
|
||||
mut session: PtySession,
|
||||
commands: &Receiver<RuntimeCommand>,
|
||||
events: &SyncSender<RuntimeEvent>,
|
||||
poll_interval: Duration,
|
||||
) {
|
||||
if send_event(
|
||||
events,
|
||||
RuntimeEvent::Started {
|
||||
session_id,
|
||||
process_id: session.process_id(),
|
||||
},
|
||||
)
|
||||
.is_err()
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
let mut output_sequence = 0_u64;
|
||||
loop {
|
||||
loop {
|
||||
match commands.try_recv() {
|
||||
Ok(command) => {
|
||||
if !handle_command(session_id, &mut session, events, command) {
|
||||
return;
|
||||
}
|
||||
}
|
||||
Err(TryRecvError::Empty) => break,
|
||||
Err(TryRecvError::Disconnected) => return,
|
||||
}
|
||||
}
|
||||
|
||||
match session.recv_output_timeout(poll_interval) {
|
||||
Ok(OutputEvent::Data(bytes)) => {
|
||||
output_sequence += 1;
|
||||
if send_event(
|
||||
events,
|
||||
RuntimeEvent::Output {
|
||||
session_id,
|
||||
sequence: output_sequence,
|
||||
bytes,
|
||||
},
|
||||
)
|
||||
.is_err()
|
||||
{
|
||||
return;
|
||||
}
|
||||
}
|
||||
Ok(OutputEvent::Eof) => match session.wait() {
|
||||
Ok(status) => {
|
||||
let _ = send_event(events, RuntimeEvent::Exited { session_id, status });
|
||||
return;
|
||||
}
|
||||
Err(error) => {
|
||||
let _ = send_fault(events, session_id, RuntimeOperation::Wait, &error);
|
||||
return;
|
||||
}
|
||||
},
|
||||
Ok(OutputEvent::ReadFailed(error)) => {
|
||||
let _ = send_event(
|
||||
events,
|
||||
RuntimeEvent::Fault {
|
||||
session_id,
|
||||
operation: RuntimeOperation::Output,
|
||||
message: error.message,
|
||||
},
|
||||
);
|
||||
return;
|
||||
}
|
||||
Err(PtyError::OutputTimeout) => match session.try_wait() {
|
||||
Ok(Some(status)) => {
|
||||
let _ = send_event(events, RuntimeEvent::Exited { session_id, status });
|
||||
return;
|
||||
}
|
||||
Ok(None) => {}
|
||||
Err(error) => {
|
||||
let _ = send_fault(events, session_id, RuntimeOperation::Wait, &error);
|
||||
return;
|
||||
}
|
||||
},
|
||||
Err(error) => {
|
||||
let _ = send_fault(events, session_id, RuntimeOperation::Output, &error);
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn handle_command(
|
||||
session_id: RuntimeSessionId,
|
||||
session: &mut PtySession,
|
||||
events: &SyncSender<RuntimeEvent>,
|
||||
command: RuntimeCommand,
|
||||
) -> bool {
|
||||
let result = match command {
|
||||
RuntimeCommand::Input(bytes) => session
|
||||
.write_all(&bytes)
|
||||
.map_err(|error| (RuntimeOperation::Input, error)),
|
||||
RuntimeCommand::Resize(size) => session
|
||||
.resize(size)
|
||||
.map_err(|error| (RuntimeOperation::Resize, error)),
|
||||
RuntimeCommand::CloseInput => {
|
||||
session.close_input();
|
||||
return send_event(events, RuntimeEvent::InputClosed { session_id }).is_ok();
|
||||
}
|
||||
RuntimeCommand::Shutdown => {
|
||||
match session.terminate() {
|
||||
Ok(status) => {
|
||||
let _ = send_event(events, RuntimeEvent::Exited { session_id, status });
|
||||
}
|
||||
Err(error) => {
|
||||
let _ = send_fault(events, session_id, RuntimeOperation::Shutdown, &error);
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
};
|
||||
|
||||
if let Err((operation, error)) = result {
|
||||
return send_fault(events, session_id, operation, &error).is_ok();
|
||||
}
|
||||
true
|
||||
}
|
||||
|
||||
fn send_event(
|
||||
events: &SyncSender<RuntimeEvent>,
|
||||
event: RuntimeEvent,
|
||||
) -> Result<(), mpsc::SendError<RuntimeEvent>> {
|
||||
events.send(event)
|
||||
}
|
||||
|
||||
fn send_fault(
|
||||
events: &SyncSender<RuntimeEvent>,
|
||||
session_id: RuntimeSessionId,
|
||||
operation: RuntimeOperation,
|
||||
error: &PtyError,
|
||||
) -> Result<(), mpsc::SendError<RuntimeEvent>> {
|
||||
send_event(
|
||||
events,
|
||||
RuntimeEvent::Fault {
|
||||
session_id,
|
||||
operation,
|
||||
message: error.to_string(),
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{
|
||||
RuntimeActor, RuntimeActorError, RuntimeActorOptions, RuntimeCommand, RuntimeEvent,
|
||||
};
|
||||
use lumbridge_pty::{CommandConfig, PtyOptions, TerminalSize};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
const TEST_TIMEOUT: Duration = Duration::from_secs(3);
|
||||
|
||||
fn shell(script: &str) -> CommandConfig {
|
||||
CommandConfig::new("/bin/sh")
|
||||
.expect("valid synthetic shell")
|
||||
.args(["-c", script])
|
||||
}
|
||||
|
||||
fn actor(script: &str) -> RuntimeActor {
|
||||
RuntimeActor::spawn(
|
||||
shell(script),
|
||||
PtyOptions::default(),
|
||||
RuntimeActorOptions::default(),
|
||||
)
|
||||
.expect("spawn runtime actor")
|
||||
}
|
||||
|
||||
fn events_until_exit(actor: &RuntimeActor) -> Vec<RuntimeEvent> {
|
||||
let deadline = Instant::now() + TEST_TIMEOUT;
|
||||
let mut events = Vec::new();
|
||||
loop {
|
||||
let remaining = deadline.saturating_duration_since(Instant::now());
|
||||
assert!(!remaining.is_zero(), "timed out waiting for actor exit");
|
||||
let event = actor.recv_timeout(remaining).expect("runtime event");
|
||||
let exited = matches!(event, RuntimeEvent::Exited { .. });
|
||||
events.push(event);
|
||||
if exited {
|
||||
return events;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn emits_started_ordered_output_and_exit() {
|
||||
let actor = actor("printf 'first\\n'; printf 'second\\n'; exit 7");
|
||||
let session_id = actor.session_id();
|
||||
let events = events_until_exit(&actor);
|
||||
|
||||
assert!(matches!(
|
||||
events.first(),
|
||||
Some(RuntimeEvent::Started {
|
||||
session_id: started,
|
||||
..
|
||||
}) if *started == session_id
|
||||
));
|
||||
let output = events
|
||||
.iter()
|
||||
.filter_map(|event| match event {
|
||||
RuntimeEvent::Output {
|
||||
sequence, bytes, ..
|
||||
} => Some((*sequence, bytes.as_slice())),
|
||||
_ => None,
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
assert!(!output.is_empty());
|
||||
assert!(
|
||||
output.windows(2).all(|pair| pair[0].0 < pair[1].0),
|
||||
"output sequence must increase"
|
||||
);
|
||||
let bytes = output
|
||||
.iter()
|
||||
.flat_map(|(_, bytes)| bytes.iter().copied())
|
||||
.collect::<Vec<_>>();
|
||||
let text = String::from_utf8_lossy(&bytes).replace('\r', "");
|
||||
assert!(
|
||||
text.contains("first\nsecond\n"),
|
||||
"unexpected output: {text:?}"
|
||||
);
|
||||
assert!(matches!(
|
||||
events.last(),
|
||||
Some(RuntimeEvent::Exited { status, .. }) if status.code == 7
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn serializes_input_and_resize_through_commands() {
|
||||
let actor = actor("IFS= read -r line; stty size; printf 'got:%s\\n' \"$line\"");
|
||||
actor
|
||||
.try_send(RuntimeCommand::Resize(
|
||||
TerminalSize::new(41, 101).expect("valid size"),
|
||||
))
|
||||
.expect("queue resize");
|
||||
actor
|
||||
.try_send(RuntimeCommand::Input(b"hello-actor\n".to_vec()))
|
||||
.expect("queue input");
|
||||
|
||||
let events = events_until_exit(&actor);
|
||||
let bytes = events
|
||||
.iter()
|
||||
.filter_map(|event| match event {
|
||||
RuntimeEvent::Output { bytes, .. } => Some(bytes.as_slice()),
|
||||
_ => None,
|
||||
})
|
||||
.flatten()
|
||||
.copied()
|
||||
.collect::<Vec<_>>();
|
||||
let text = String::from_utf8_lossy(&bytes).replace('\r', "");
|
||||
assert!(text.contains("41 101"), "unexpected output: {text:?}");
|
||||
assert!(
|
||||
text.contains("got:hello-actor"),
|
||||
"unexpected output: {text:?}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn drop_terminates_a_hung_session_without_waiting_for_output() {
|
||||
let actor = actor("trap '' HUP; exec sleep 30");
|
||||
let process_id = actor.process_id().expect("Unix shell has a process id");
|
||||
let started = Instant::now();
|
||||
drop(actor);
|
||||
assert!(started.elapsed() < TEST_TIMEOUT);
|
||||
|
||||
#[cfg(target_os = "linux")]
|
||||
{
|
||||
let proc_path = format!("/proc/{process_id}");
|
||||
assert!(!std::path::Path::new(&proc_path).exists());
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn reports_timeout_without_exposing_command_contents() {
|
||||
let actor = actor("sleep 1");
|
||||
assert!(matches!(
|
||||
actor.recv_timeout(Duration::ZERO),
|
||||
Err(RuntimeActorError::EventTimeout) | Ok(RuntimeEvent::Started { .. })
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_a_zero_poll_interval_before_spawning() {
|
||||
let options = RuntimeActorOptions {
|
||||
poll_interval: Duration::ZERO,
|
||||
..RuntimeActorOptions::default()
|
||||
};
|
||||
let result = RuntimeActor::spawn(shell("exit 0"), PtyOptions::default(), options);
|
||||
assert!(matches!(
|
||||
result,
|
||||
Err(RuntimeActorError::InvalidPollInterval)
|
||||
));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user