feat(q): decode Q wire messages received by an external transport

rayforce-sys/src/lib.rs

@@ -38,6 +38,17 @@ extern "C" {
         err: *mut ::std::os::raw::c_char,
         errlen: usize,
     ) -> *mut ray_t;
+    /// Decompress (if needed) and deserialize a response *body* (wire header
+    /// already stripped) into a freshly-owned object — possibly a `RAY_ERROR`
+    /// carrying a Q server-side error. Touches the symbol table: engine thread
+    /// only. Returns null on a decode failure with a short reason in `err`.
+    pub fn q_decode(
+        resp: *mut u8,
+        resp_len: i64,
+        compressed: ::std::os::raw::c_int,
+        err: *mut ::std::os::raw::c_char,
+        errlen: usize,
+    ) -> *mut ray_t;
 }
 
 /// `q_connect` failure codes (mirrors `q.h`).

rayforce/src/q.rs

@@ -94,3 +94,53 @@ impl Drop for QConnection {
         unsafe { sys::q_close(self.fd) };
     }
 }
+
+/// Decode a complete Q response message (8-byte wire header + body) that was
+/// received by an **external transport** — e.g. a worker thread that owns a
+/// plain `std::net::TcpStream` with read timeouts — into a [`Value`].
+///
+/// This is the engine-thread half of a split client: socket I/O is plain
+/// bytes and thread-safe anywhere, while this call allocates engine objects
+/// and must run on the thread that owns the [`crate::Runtime`] (like every
+/// other constructor in this crate). A Q server-side error surfaces as `Err`.
+pub fn decode_response(msg: &[u8]) -> Result<Value> {
+    // q_header_t (q.c): endianness, msgtype, compressed, reserved, u32 size.
+    // `size` counts the whole message, header included. Little-endian wire only.
+    const HEADER_LEN: usize = 8;
+    if msg.len() < HEADER_LEN {
+        return Err(RayError::binding("Q: message shorter than wire header"));
+    }
+    let size = u32::from_le_bytes([msg[4], msg[5], msg[6], msg[7]]) as usize;
+    if size != msg.len() {
+        return Err(RayError::binding(
+            "Q: message length does not match wire header",
+        ));
+    }
+    let compressed = i32::from(msg[2] != 0);
+    let body = &msg[HEADER_LEN..];
+    if body.is_empty() {
+        return Err(RayError::binding("Q: empty response body"));
+    }
+
+    let mut err = [0i8; 256];
+    let res = unsafe {
+        sys::q_decode(
+            body.as_ptr() as *mut u8,
+            body.len() as i64,
+            compressed,
+            err.as_mut_ptr(),
+            err.len(),
+        )
+    };
+    if res.is_null() {
+        let msg = unsafe { std::ffi::CStr::from_ptr(err.as_ptr()) }.to_string_lossy();
+        let msg = if msg.is_empty() {
+            "q_decode failed".into()
+        } else {
+            msg
+        };
+        return Err(RayError::binding(format!("Q: {msg}")));
+    }
+    let ok = unsafe { check(res)? };
+    Ok(unsafe { Value::from_owned(ok) })
+}