aboutsummaryrefslogtreecommitdiffhomepage
path: root/crates/shirabe-php-rpc/src/lib.rs
diff options
context:
space:
mode:
Diffstat (limited to 'crates/shirabe-php-rpc/src/lib.rs')
-rw-r--r--crates/shirabe-php-rpc/src/lib.rs547
1 files changed, 256 insertions, 291 deletions
diff --git a/crates/shirabe-php-rpc/src/lib.rs b/crates/shirabe-php-rpc/src/lib.rs
index 9b9c9222..491af072 100644
--- a/crates/shirabe-php-rpc/src/lib.rs
+++ b/crates/shirabe-php-rpc/src/lib.rs
@@ -1,11 +1,17 @@
//! Rust-to-PHP RPC over a Unix domain socket. See `docs/dev/php-rpc.md`.
-use anyhow::Context as _;
+pub mod frame;
+pub mod session;
+pub mod value;
+
+pub use value::{PhpClassHandle, PhpObjHandle, PluginValue, RustObjHandle};
+
+use frame::Frame;
use indexmap::IndexMap;
use shirabe_external_packages::symfony::process::PhpExecutableFinder;
use shirabe_php_shim::PhpMixed;
-use std::io::{Read as _, Write as _};
use std::os::unix::net::{UnixListener, UnixStream};
+use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{LazyLock, Mutex, OnceLock};
use std::time::{Duration, Instant};
@@ -294,8 +300,207 @@ fn string_list(value: PhpMixed, name: &str) -> Vec<String> {
}
}
+/// A PHP exception that crossed the RPC boundary (the recoverable failure lane, as opposed to
+/// the fatal `anyhow::Error` lane used for a dead worker or a broken channel).
+#[derive(Debug, Clone)]
+pub struct PhpThrow {
+ pub exception_class: String,
+ pub message: String,
+ pub code: i64,
+}
+
+impl PhpThrow {
+ fn runtime(message: String) -> PhpThrow {
+ PhpThrow {
+ exception_class: "RuntimeException".to_string(),
+ message,
+ code: 0,
+ }
+ }
+}
+
+impl std::fmt::Display for PhpThrow {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ write!(f, "{}: {}", self.exception_class, self.message)
+ }
+}
+
+impl std::error::Error for PhpThrow {}
+
+/// Handles `CallRustMethod` requests arriving while a Rust-initiated call is waiting for its
+/// `Return` (the cooperative reentrancy loop). A handler may itself issue nested RPC calls.
+pub trait RustMethodDispatcher {
+ fn dispatch(
+ &mut self,
+ rhandle: u64,
+ method_name: &str,
+ args: Vec<PluginValue>,
+ out_param_positions: &[u32],
+ ) -> Result<PluginValue, PhpThrow>;
+}
+
+/// The Rust side allocates odd correlation ids; the PHP side allocates even ones. Calls nest
+/// strictly, so this split is not needed for disambiguation — it just makes any captured frame
+/// attributable to its initiator.
+static NEXT_CORR_ID: AtomicU64 = AtomicU64::new(1);
+
+/// Allocates a Rust-side object handle. Handle 0 is reserved for the runtime service endpoint
+/// (e.g. `__shirabe_find_file` autoload queries), so ids start at 1.
+static NEXT_RHANDLE: AtomicU64 = AtomicU64::new(1);
+
+pub fn alloc_rhandle() -> u64 {
+ NEXT_RHANDLE.fetch_add(1, Ordering::Relaxed)
+}
+
+/// Calls a PHP function in the worker. The outer `Result` is the fatal lane (dead worker, broken
+/// framing); the inner one carries a PHP exception if the call threw.
+pub fn call_function(
+ name: &str,
+ args: Vec<PluginValue>,
+) -> anyhow::Result<Result<PluginValue, PhpThrow>> {
+ call_function_with_dispatcher(name, args, None)
+}
+
+pub fn call_function_with_dispatcher(
+ name: &str,
+ args: Vec<PluginValue>,
+ dispatcher: Option<&mut dyn RustMethodDispatcher>,
+) -> anyhow::Result<Result<PluginValue, PhpThrow>> {
+ rpc_call(
+ |corr_id| Frame::CallFunction {
+ corr_id,
+ function_name: name.to_string(),
+ args,
+ out_param_positions: Vec::new(),
+ },
+ dispatcher,
+ )
+}
+
+/// Calls `$class::$method(...$args)` in the worker (autoloading the class if needed).
+pub fn call_static_method(
+ class: &str,
+ method: &str,
+ args: Vec<PluginValue>,
+ dispatcher: Option<&mut dyn RustMethodDispatcher>,
+) -> anyhow::Result<Result<PluginValue, PhpThrow>> {
+ rpc_call(
+ |corr_id| Frame::CallStaticMethod {
+ corr_id,
+ pclass: class.to_string(),
+ method_name: method.to_string(),
+ args,
+ out_param_positions: Vec::new(),
+ },
+ dispatcher,
+ )
+}
+
+fn rpc_call(
+ request: impl FnOnce(u64) -> Frame,
+ mut dispatcher: Option<&mut dyn RustMethodDispatcher>,
+) -> anyhow::Result<Result<PluginValue, PhpThrow>> {
+ // Held for the whole logical call session; nested calls from the same thread (issued by a
+ // dispatcher handler) re-enter immediately, other threads are serialized.
+ let _session = session::SessionGuard::enter();
+ let my_id = NEXT_CORR_ID.fetch_add(2, Ordering::Relaxed);
+ send_frame(&request(my_id))?;
+ loop {
+ let incoming = recv_frame()?;
+ match incoming {
+ Frame::Return { corr_id, value, .. } if corr_id == my_id => {
+ return Ok(Ok(value));
+ }
+ Frame::Throw {
+ corr_id,
+ exception_class,
+ message,
+ code,
+ } if corr_id == my_id => {
+ return Ok(Err(PhpThrow {
+ exception_class,
+ message,
+ code,
+ }));
+ }
+ Frame::CallRustMethod {
+ corr_id,
+ rhandle,
+ method_name,
+ args,
+ out_param_positions,
+ } => {
+ let outcome = match dispatcher.as_deref_mut() {
+ Some(dispatcher) => {
+ dispatcher.dispatch(rhandle, &method_name, args, &out_param_positions)
+ }
+ // Never fall back to a silent null: an unroutable callback is reported as an
+ // explicit error on the PHP side.
+ None => Err(PhpThrow::runtime(format!(
+ "no Rust method dispatcher is active for this call \
+ (rhandle {rhandle}, method `{method_name}`)"
+ ))),
+ };
+ let reply = match outcome {
+ Ok(value) => Frame::Return {
+ corr_id,
+ value,
+ out_params: IndexMap::new(),
+ },
+ Err(throw) => Frame::Throw {
+ corr_id,
+ exception_class: throw.exception_class,
+ message: throw.message,
+ code: throw.code,
+ },
+ };
+ send_frame(&reply)?;
+ }
+ Frame::ReleaseRustHandle { .. } => {
+ // TODO(plugin): there is no persistent R table yet (script-event handles are
+ // scoped to a single dispatched call), so stub destructor notifications carry no
+ // state to clean up.
+ continue;
+ }
+ Frame::EpochBump { .. } => {
+ // Rust is the sender of epoch bumps; tolerate the symmetric direction.
+ continue;
+ }
+ other => panic!(
+ "PHP RPC: protocol violation — unexpected frame while waiting for corr_id \
+ {my_id}: {other:?}"
+ ),
+ }
+ }
+}
+
+fn call(name: &str, arg: &str) -> PhpMixed {
+ let outcome = call_function(name, vec![PluginValue::string(arg)])
+ .unwrap_or_else(|e| panic!("PHP RPC: request `{name}` failed: {e:#}"));
+ let value = match outcome {
+ Ok(value) => value,
+ Err(throw) => panic!("PHP RPC: request `{name}` threw {throw}"),
+ };
+ value
+ .to_php_mixed()
+ .unwrap_or_else(|e| panic!("PHP RPC: request `{name}` returned an unusable value: {e:#}"))
+}
+
const GLUE_SCRIPT: &str = include_str!("../php/worker.php");
+/// Hand-written proxy stub classes made autoloadable inside the worker, written in the shape the
+/// future stub generator will output.
+const STUB_FILES: &[(&str, &str)] = &[
+ (
+ "Composer/EventDispatcher/Event.php",
+ include_str!("../php/stubs/Composer/EventDispatcher/Event.php"),
+ ),
+ (
+ "Composer/Script/Event.php",
+ include_str!("../php/stubs/Composer/Script/Event.php"),
+ ),
+];
+
struct Worker {
stream: UnixStream,
// Also queried for its exit status when a socket read/write fails, to tell a dead worker
@@ -307,14 +512,6 @@ struct Worker {
}
impl Worker {
- fn request(&mut self, name: &str, arg: &str) -> anyhow::Result<Vec<u8>> {
- let mut payload = name.as_bytes().to_vec();
- payload.push(0);
- payload.extend_from_slice(arg.as_bytes());
- write_frame(&mut self.stream, &payload).with_context(|| self.worker_state())?;
- read_frame(&mut self.stream).with_context(|| self.worker_state())
- }
-
/// Describes the PHP worker's current process state, to be attached as `anyhow::Context` to
/// an I/O error so a dead worker (crash, OOM kill, ...) can be told apart from a live one
/// hitting a framing bug.
@@ -330,7 +527,7 @@ impl Worker {
}
}
-// TODO(phase-c): every failure here panics rather than propagating a `Result`; this is an interim
+// TODO(phase-c): a failed spawn panics rather than propagating a `Result`; this is an interim
// step until PHP RPC gets proper error handling (see docs/dev/php-rpc.md).
static WORKER: LazyLock<Mutex<Worker>> = LazyLock::new(|| {
Mutex::new(
@@ -338,16 +535,23 @@ static WORKER: LazyLock<Mutex<Worker>> = LazyLock::new(|| {
)
});
-fn call(name: &str, arg: &str) -> PhpMixed {
+/// Writes one frame while holding the worker mutex only for the duration of the write, so the
+/// session owner (see `session`) can interleave sends and blocking reads without keeping the
+/// worker locked across a whole call.
+fn send_frame(frame: &Frame) -> anyhow::Result<()> {
let mut guard = WORKER
.lock()
.unwrap_or_else(|e| panic!("PHP RPC: worker mutex poisoned: {e}"));
- let payload = guard
- .request(name, arg)
- .unwrap_or_else(|e| panic!("PHP RPC: request `{name}` failed: {e:#}"));
- parse_serialized_value(&payload).unwrap_or_else(|| {
- panic!("PHP RPC: request `{name}` returned an unparseable payload: {payload:?}")
- })
+ let result = frame::write_frame(&mut guard.stream, frame);
+ result.map_err(|e| anyhow::Error::new(e).context(guard.worker_state()))
+}
+
+fn recv_frame() -> anyhow::Result<Frame> {
+ let mut guard = WORKER
+ .lock()
+ .unwrap_or_else(|e| panic!("PHP RPC: worker mutex poisoned: {e}"));
+ let result = frame::read_frame(&mut guard.stream);
+ result.map_err(|e| anyhow::Error::new(e).context(guard.worker_state()))
}
fn spawn_worker() -> anyhow::Result<Worker> {
@@ -360,13 +564,33 @@ fn spawn_worker() -> anyhow::Result<Worker> {
let script_path = tempdir.path().join("worker.php");
std::fs::write(&script_path, GLUE_SCRIPT)?;
+ let stubs_dir = tempdir.path().join("stubs");
+ for (relative_path, contents) in STUB_FILES {
+ let path = stubs_dir.join(relative_path);
+ std::fs::create_dir_all(path.parent().expect("stub paths have a parent"))?;
+ std::fs::write(&path, contents)?;
+ }
+
// Bind before spawning so the socket exists when the child connects.
let listener = UnixListener::bind(&socket_path)?;
listener.set_nonblocking(true)?;
+ // The socket lives in a 0700 temp dir already; restricting the socket file itself makes the
+ // protection independent of the directory permission.
+ std::fs::set_permissions(
+ &socket_path,
+ std::os::unix::fs::PermissionsExt::from_mode(0o600),
+ )?;
+
let child = std::process::Command::new(&php)
+ // The Rust-side codec produces the byte representation of the default (and only
+ // supported) serialize_precision; pin the child to it in case a distro php.ini overrides
+ // the default.
+ .arg("-d")
+ .arg("serialize_precision=-1")
.arg(&script_path)
.arg(&socket_path)
+ .arg(&stubs_dir)
.spawn()?;
// Poll for the child's connection with a bounded deadline so a child that never connects does
@@ -393,280 +617,35 @@ fn spawn_worker() -> anyhow::Result<Worker> {
})
}
-fn write_frame(stream: &mut UnixStream, payload: &[u8]) -> std::io::Result<()> {
- stream.write_all(&(payload.len() as u64).to_le_bytes())?;
- stream.write_all(payload)?;
- stream.flush()
-}
-
-fn read_frame(stream: &mut UnixStream) -> std::io::Result<Vec<u8>> {
- let mut header = [0u8; 8];
- stream.read_exact(&mut header)?;
- let len = u64::from_le_bytes(header) as usize;
- let mut payload = vec![0u8; len];
- stream.read_exact(&mut payload)?;
- Ok(payload)
-}
-
-/// Parse a whole `serialize()` payload, rejecting trailing garbage.
-fn parse_serialized_value(payload: &[u8]) -> Option<PhpMixed> {
- let mut pos = 0;
- let value = parse_value(payload, &mut pos)?;
- (pos == payload.len()).then_some(value)
-}
-
-/// Parse one `serialize()` value starting at `pos`, advancing it past the value: `N;`, `b:0/1;`,
-/// `i:<n>;`, `d:<f>;`, `s:<len>:"<bytes>";`, `a:<count>:{<key><value>...}`.
-fn parse_value(payload: &[u8], pos: &mut usize) -> Option<PhpMixed> {
- let tag = payload.get(*pos..*pos + 2)?;
- *pos += 2;
- match tag {
- b"N;" => Some(PhpMixed::Null),
- b"b:" => match take_until(payload, pos, b';')? {
- b"0" => Some(PhpMixed::Bool(false)),
- b"1" => Some(PhpMixed::Bool(true)),
- _ => None,
- },
- b"i:" => parse_int(take_until(payload, pos, b';')?).map(PhpMixed::Int),
- b"d:" => std::str::from_utf8(take_until(payload, pos, b';')?)
- .ok()?
- .parse()
- .ok()
- .map(PhpMixed::Float),
- b"s:" => parse_string_body(payload, pos).map(PhpMixed::String),
- b"a:" => parse_array_body(payload, pos),
- _ => None,
- }
-}
-
-/// Parse the `<len>:"<bytes>";` tail of a serialized string.
-fn parse_string_body(payload: &[u8], pos: &mut usize) -> Option<String> {
- let len: usize = std::str::from_utf8(take_until(payload, pos, b':')?)
- .ok()?
- .parse()
- .ok()?;
- if payload.get(*pos) != Some(&b'"') {
- return None;
- }
- *pos += 1;
- let bytes = payload.get(*pos..*pos + len)?;
- *pos += len;
- if payload.get(*pos..*pos + 2) != Some(b"\";") {
- return None;
- }
- *pos += 2;
- Some(String::from_utf8_lossy(bytes).into_owned())
-}
-
-/// Parse the `<count>:{<key><value>...}` tail of a serialized array. An array whose keys are
-/// exactly `0..count` maps to `PhpMixed::List`, matching how PHP renders such an array as a JSON
-/// list; anything else maps to `PhpMixed::Array` with the keys stringified.
-fn parse_array_body(payload: &[u8], pos: &mut usize) -> Option<PhpMixed> {
- let count: usize = std::str::from_utf8(take_until(payload, pos, b':')?)
- .ok()?
- .parse()
- .ok()?;
- if payload.get(*pos) != Some(&b'{') {
- return None;
- }
- *pos += 1;
-
- let mut entries: IndexMap<String, PhpMixed> = IndexMap::new();
- let mut is_list = true;
- for index in 0..count {
- let key = match parse_value(payload, pos)? {
- PhpMixed::Int(n) => {
- is_list &= n == index as i64;
- n.to_string()
- }
- PhpMixed::String(s) => {
- is_list = false;
- s
- }
- _ => return None,
- };
- entries.insert(key, parse_value(payload, pos)?);
- }
-
- if payload.get(*pos) != Some(&b'}') {
- return None;
- }
- *pos += 1;
-
- Some(if is_list {
- PhpMixed::List(entries.into_values().collect())
- } else {
- PhpMixed::Array(entries)
- })
-}
-
-/// Return the bytes from `pos` up to the next `terminator`, advancing `pos` past it.
-fn take_until<'a>(payload: &'a [u8], pos: &mut usize, terminator: u8) -> Option<&'a [u8]> {
- let end = *pos + payload.get(*pos..)?.iter().position(|&b| b == terminator)?;
- let bytes = &payload[*pos..end];
- *pos = end + 1;
- Some(bytes)
-}
-
-fn parse_int(bytes: &[u8]) -> Option<i64> {
- std::str::from_utf8(bytes).ok()?.parse().ok()
-}
-
#[cfg(test)]
mod tests {
use super::*;
#[test]
- fn parses_string_scalar() {
- assert_eq!(
- parse_serialized_value(b"s:5:\"8.5.7\";"),
- Some(PhpMixed::String("8.5.7".to_string())),
- );
- }
-
- #[test]
- fn parses_empty_string() {
- assert_eq!(
- parse_serialized_value(b"s:0:\"\";"),
- Some(PhpMixed::String(String::new())),
- );
- }
-
- #[test]
- fn parses_string_with_embedded_quote() {
- assert_eq!(
- parse_serialized_value(b"s:3:\"a\"b\";"),
- Some(PhpMixed::String("a\"b".to_string())),
- );
- }
-
- #[test]
- fn rejects_truncated_string() {
- assert_eq!(parse_serialized_value(b"s:5:\"ab\";"), None);
- }
-
- #[test]
- fn rejects_trailing_garbage() {
- assert_eq!(parse_serialized_value(b"i:42;i:43;"), None);
- }
-
- #[test]
fn request_error_reports_dead_worker_exit_status() {
let mut worker = spawn_worker().expect("failed to spawn PHP worker");
worker.child.kill().expect("failed to kill PHP worker");
worker.child.wait().expect("failed to reap PHP worker");
- let err = worker
- .request("defined", "PHP_VERSION")
- .expect_err("request against a dead worker should fail");
- let message = format!("{err:#}");
- assert!(
- message.contains("PHP worker process already exited"),
- "unexpected error message: {message}"
+ // Writing may still succeed into the socket buffer; the read is what must fail.
+ let _ = frame::write_frame(
+ &mut worker.stream,
+ &Frame::CallFunction {
+ corr_id: 1,
+ function_name: "defined".to_string(),
+ args: vec![PluginValue::string("PHP_VERSION")],
+ out_param_positions: Vec::new(),
+ },
);
- }
-
- #[test]
- fn rejects_non_numeric_length() {
- assert_eq!(parse_serialized_value(b"s:x:\"ab\";"), None);
- }
-
- #[test]
- fn parses_scalar_null() {
- assert_eq!(parse_serialized_value(b"N;"), Some(PhpMixed::Null));
- }
-
- #[test]
- fn parses_scalar_bool() {
- assert_eq!(parse_serialized_value(b"b:0;"), Some(PhpMixed::Bool(false)));
- assert_eq!(parse_serialized_value(b"b:1;"), Some(PhpMixed::Bool(true)));
- }
-
- #[test]
- fn parses_scalar_int() {
- assert_eq!(parse_serialized_value(b"i:8;"), Some(PhpMixed::Int(8)));
- assert_eq!(parse_serialized_value(b"i:-1;"), Some(PhpMixed::Int(-1)));
- }
-
- #[test]
- fn parses_scalar_float() {
- assert_eq!(
- parse_serialized_value(b"d:1.5;"),
- Some(PhpMixed::Float(1.5))
- );
- }
-
- #[test]
- fn rejects_malformed_scalar() {
- assert_eq!(parse_serialized_value(b"b:2;"), None);
- assert_eq!(parse_serialized_value(b"i:x;"), None);
- assert_eq!(parse_serialized_value(b"d:x;"), None);
- assert_eq!(parse_serialized_value(b"garbage"), None);
- }
-
- #[test]
- fn parses_list_array() {
- assert_eq!(
- parse_serialized_value(b"a:2:{i:0;s:1:\"a\";i:1;i:7;}"),
- Some(PhpMixed::List(vec![
- PhpMixed::String("a".to_string()),
- PhpMixed::Int(7),
- ])),
- );
- }
-
- #[test]
- fn parses_empty_array_as_list() {
- assert_eq!(
- parse_serialized_value(b"a:0:{}"),
- Some(PhpMixed::List(vec![]))
- );
- }
-
- #[test]
- fn parses_keyed_array() {
- let expected: IndexMap<String, PhpMixed> = [
- ("zip".to_string(), PhpMixed::Bool(true)),
- ("apcu".to_string(), PhpMixed::Null),
- ]
- .into_iter()
- .collect();
- assert_eq!(
- parse_serialized_value(b"a:2:{s:3:\"zip\";b:1;s:4:\"apcu\";N;}"),
- Some(PhpMixed::Array(expected)),
- );
- }
-
- #[test]
- fn parses_nested_array() {
- let inner: IndexMap<String, PhpMixed> = [("curl".to_string(), PhpMixed::Bool(false))]
- .into_iter()
- .collect();
- let expected: IndexMap<String, PhpMixed> = [
- ("extensions".to_string(), PhpMixed::Array(inner)),
- ("php_version_id".to_string(), PhpMixed::Int(80500)),
- ]
- .into_iter()
- .collect();
- assert_eq!(
- parse_serialized_value(
- b"a:2:{s:10:\"extensions\";a:1:{s:4:\"curl\";b:0;}s:14:\"php_version_id\";i:80500;}"
- ),
- Some(PhpMixed::Array(expected)),
+ frame::read_frame(&mut worker.stream).expect_err("reading from a dead worker should fail");
+ let state = worker.worker_state();
+ assert!(
+ state.contains("PHP worker process already exited"),
+ "unexpected worker state: {state}"
);
}
#[test]
- fn rejects_malformed_array() {
- // Count larger than the number of entries.
- assert_eq!(parse_serialized_value(b"a:2:{i:0;i:1;}"), None);
- // Missing closing brace.
- assert_eq!(parse_serialized_value(b"a:1:{i:0;i:1;"), None);
- // Non-scalar key.
- assert_eq!(parse_serialized_value(b"a:1:{N;i:1;}"), None);
- }
-
- #[test]
fn queries_string_lists_when_php_available() {
if PhpExecutableFinder::new().find(false).is_none() {
// No PHP in this environment; the worker cannot start.
@@ -706,20 +685,6 @@ mod tests {
}
#[test]
- fn frame_roundtrip() {
- let (mut a, mut b) = UnixStream::pair().unwrap();
- write_frame(&mut a, b"get_php_version").unwrap();
- assert_eq!(read_frame(&mut b).unwrap(), b"get_php_version");
- }
-
- #[test]
- fn frame_roundtrip_empty_payload() {
- let (mut a, mut b) = UnixStream::pair().unwrap();
- write_frame(&mut a, b"").unwrap();
- assert_eq!(read_frame(&mut b).unwrap(), b"");
- }
-
- #[test]
fn queries_real_php_when_available() {
if PhpExecutableFinder::new().find(false).is_none() {
// No PHP in this environment; the worker cannot start.
generated by cgit v1.3.1 (git 2.54.0) at 2026-09-04 15:02:41 +0000