use super::{framing::read_frame, policy};
use crate::{Error, Result};
use serde_json::{json, Value};
use std::{
collections::{BTreeMap, BTreeSet, HashMap},
path::{Path, PathBuf},
process::Stdio,
sync::{
atomic::{AtomicBool, AtomicU8, Ordering},
Arc, Mutex, OnceLock,
},
time::Duration,
};
use tokio::{
io::{AsyncReadExt, AsyncWriteExt, BufReader},
process::{Child, ChildStdin, Command},
sync::{broadcast, mpsc, oneshot, watch, Notify, OwnedSemaphorePermit, Semaphore},
time::{timeout, Instant},
};
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub enum HostMode {
#[default]
Cli,
Document,
}
#[derive(Clone)]
pub struct HostOptions {
pub executable: PathBuf,
pub prefix_args: Vec<String>,
pub mode: HostMode,
pub env: BTreeMap<String, String>,
pub startup_timeout: Duration,
pub request_timeout: Duration,
pub shutdown_timeout: Duration,
pub max_pending: usize,
pub max_frame_bytes: usize,
pub state_directory: Option<PathBuf>,
}
impl HostOptions {
pub fn new(executable: impl Into<PathBuf>) -> Self {
let default_ms = policy()["hostDefaultRequestTimeoutMs"]
.as_u64()
.unwrap_or(30_000);
Self {
executable: executable.into(),
prefix_args: Vec::new(),
mode: HostMode::Cli,
env: BTreeMap::new(),
startup_timeout: Duration::from_secs(10),
request_timeout: Duration::from_millis(default_ms),
shutdown_timeout: Duration::from_secs(5),
max_pending: 64,
max_frame_bytes: 64 * 1024 * 1024,
state_directory: None,
}
}
}
impl std::fmt::Debug for HostOptions {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter
.debug_struct("HostOptions")
.field("mode", &self.mode)
.field("max_pending", &self.max_pending)
.field("max_frame_bytes", &self.max_frame_bytes)
.finish_non_exhaustive()
}
}
#[derive(Default)]
pub(crate) struct CallState {
phase: AtomicU8,
written: AtomicBool,
asset: Mutex<Option<tempfile::TempPath>>,
}
impl CallState {
pub(crate) fn with_asset(asset: tempfile::TempPath) -> Arc<Self> {
Arc::new(Self {
asset: Mutex::new(Some(asset)),
..Self::default()
})
}
pub(crate) fn cancel_before_send(&self) -> bool {
self.phase
.compare_exchange(0, 2, Ordering::AcqRel, Ordering::Acquire)
!= Err(1)
}
}
struct CancelGuard {
state: Arc<CallState>,
retire: Option<Arc<Notify>>,
disarmed: bool,
}
impl Drop for CancelGuard {
fn drop(&mut self) {
if !self.disarmed
&& self
.state
.phase
.compare_exchange(0, 2, Ordering::AcqRel, Ordering::Acquire)
== Err(1)
{
if let Some(stop) = &self.retire {
stop.notify_one();
}
}
}
}
type Sessions = Arc<Mutex<BTreeMap<String, Arc<AtomicU8>>>>;
struct Request {
method: String,
params: Value,
request_timeout: Option<Duration>,
watchdog: Duration,
mutates: bool,
abandon_open: bool,
state: Arc<CallState>,
reply: oneshot::Sender<Result<Value>>,
permit: OwnedSemaphorePermit,
}
struct Pending {
reply: oneshot::Sender<Result<Value>>,
state: Arc<CallState>,
deadline: Instant,
mutates: bool,
abandon_open: bool,
_permit: OwnedSemaphorePermit,
}
struct Write {
id: u64,
frame: Vec<u8>,
state: Arc<CallState>,
}
enum Event {
Frame(Value),
Failed(Error),
Skipped(u64),
}
enum Control {
Request(Request),
Shutdown,
}
struct Connection {
tx: mpsc::Sender<Control>,
permits: Arc<Semaphore>,
closed: Arc<AtomicBool>,
stop: Arc<Notify>,
features: OnceLock<BTreeSet<String>>,
notifications: Mutex<broadcast::Receiver<Value>>,
state_directory: PathBuf,
options: HostOptions,
termination: watch::Receiver<Option<Result<()>>>,
sessions: Sessions,
}
impl Drop for Connection {
fn drop(&mut self) {
self.stop.notify_one();
}
}
#[derive(Clone)]
pub(crate) struct Transport(Arc<Connection>);
impl Transport {
pub async fn connect(options: HostOptions) -> Result<Self> {
if options.executable.as_os_str().is_empty()
|| options.max_pending == 0
|| options.max_frame_bytes == 0
|| options.startup_timeout.is_zero()
|| options.request_timeout.is_zero()
|| options.shutdown_timeout.is_zero()
|| options.max_pending > 1024
|| [
options.startup_timeout,
options.request_timeout,
options.shutdown_timeout,
]
.iter()
.any(|duration| duration.as_millis() == 0 || *duration > Duration::from_secs(86400))
{
return Err(Error::new(
"INVALID_ARGUMENT",
"host paths, limits and timeouts must be non-empty and positive",
));
}
let (directory, owned) = match &options.state_directory {
Some(path) => {
std::fs::create_dir_all(path).map_err(|_| {
Error::new(
"STATE_DIRECTORY_ERROR",
"could not prepare the configured state directory",
)
})?;
(path.clone(), false)
}
None => (
tempfile::Builder::new()
.prefix("superdoc-rust-")
.tempdir()
.map_err(|_| {
Error::new(
"STATE_DIRECTORY_ERROR",
"could not create an SDK state directory",
)
})?
.keep(),
true,
),
};
let mut command = Command::new(&options.executable);
command
.args(&options.prefix_args)
.envs(&options.env)
.env("SUPERDOC_CLI_STATE_DIR", &directory)
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.kill_on_drop(true);
if options.mode == HostMode::Cli {
command.args(["host", "--stdio"]);
}
let mut child = match command.spawn() {
Ok(child) => child,
Err(_) => {
if owned {
let _ = std::fs::remove_dir_all(&directory);
}
return Err(Error::new(
"HOST_SPAWN_FAILED",
"could not start the configured SuperDoc host",
));
}
};
let stdin = child.stdin.take().expect("piped stdin");
let stdout = child.stdout.take().expect("piped stdout");
let stderr = child.stderr.take().expect("piped stderr");
let (tx, rx) = mpsc::channel(options.max_pending);
let (write_tx, write_rx) = mpsc::channel(options.max_pending);
let (event_tx, event_rx) = mpsc::channel(2);
let (notifications, notification_receiver) = broadcast::channel(64);
let (termination_tx, termination) = watch::channel(None);
let stop = Arc::new(Notify::new());
let closed = Arc::new(AtomicBool::new(false));
let sessions = Arc::new(Mutex::new(BTreeMap::new()));
let reader_events = event_tx.clone();
let limit = options.max_frame_bytes;
let reader = tokio::spawn(async move {
let mut input = BufReader::new(stdout);
loop {
let event = match read_frame(&mut input, limit).await {
Ok(Some(frame)) => Event::Frame(frame),
Ok(None) => {
Event::Failed(Error::new("HOST_DISCONNECTED", "host stdout closed"))
}
Err(error) => Event::Failed(error),
};
let failed = matches!(event, Event::Failed(_));
if reader_events.send(event).await.is_err() || failed {
break;
}
}
});
let writer = tokio::spawn(write_frames(stdin, write_rx, event_tx));
let stderr_reader = tokio::spawn(async move {
let mut stderr = stderr;
let mut bytes = [0u8; 4096];
while let Ok(count) = stderr.read(&mut bytes).await {
if count == 0 {
break;
}
}
});
let transport = Self(Arc::new(Connection {
tx,
permits: Arc::new(Semaphore::new(options.max_pending)),
closed: closed.clone(),
stop: stop.clone(),
features: OnceLock::new(),
notifications: Mutex::new(notification_receiver),
state_directory: directory.clone(),
options: options.clone(),
termination,
sessions: sessions.clone(),
}));
tokio::spawn(run_actor(
child,
rx,
write_tx,
event_rx,
stop,
closed,
notifications,
directory,
owned,
options.shutdown_timeout,
reader,
writer,
stderr_reader,
termination_tx,
sessions,
));
let capabilities = transport
.request(
"host.capabilities",
json!({}),
None,
options.startup_timeout,
false,
false,
)
.await;
let features = capabilities.and_then(|value| validate_capabilities(&value, options.mode));
match features {
Ok(features) => {
let _ = transport.0.features.set(features);
Ok(transport)
}
Err(error) => {
let _ = transport.shutdown().await;
Err(Error {
code: "HOST_HANDSHAKE_FAILED".to_owned(),
..error
})
}
}
}
pub fn features(&self) -> &BTreeSet<String> {
self.0.features.get().expect("completed handshake")
}
pub fn options(&self) -> &HostOptions {
&self.0.options
}
pub fn is_closed(&self) -> bool {
self.0.closed.load(Ordering::Acquire)
}
pub fn state_directory(&self) -> &Path {
&self.0.state_directory
}
pub fn subscribe(&self) -> broadcast::Receiver<Value> {
self.0
.notifications
.lock()
.expect("SDK notification lock")
.resubscribe()
}
pub fn session_state(&self, id: &str) -> Arc<AtomicU8> {
self.0
.sessions
.lock()
.expect("SDK session lock")
.entry(id.to_owned())
.or_insert_with(|| Arc::new(AtomicU8::new(0)))
.clone()
}
pub fn has_session(&self, id: &str) -> bool {
self.0
.sessions
.lock()
.expect("SDK session lock")
.get(id)
.is_some_and(|state| state.load(Ordering::Acquire) < 2)
}
pub fn remove_session(&self, id: &str, state: &Arc<AtomicU8>) {
let mut sessions = self.0.sessions.lock().expect("SDK session lock");
if sessions
.get(id)
.is_some_and(|current| Arc::ptr_eq(current, state))
{
sessions.remove(id);
}
}
pub async fn request(
&self,
method: &str,
params: Value,
request_timeout: Option<Duration>,
watchdog: Duration,
mutates: bool,
abandon_open: bool,
) -> Result<Value> {
self.request_tracked(
method,
params,
request_timeout,
watchdog,
mutates,
abandon_open,
Arc::new(CallState::default()),
)
.await
}
#[allow(clippy::too_many_arguments)]
pub(crate) async fn request_tracked(
&self,
method: &str,
params: Value,
request_timeout: Option<Duration>,
watchdog: Duration,
mutates: bool,
abandon_open: bool,
state: Arc<CallState>,
) -> Result<Value> {
for duration in [Some(watchdog), request_timeout].into_iter().flatten() {
if duration.as_millis() == 0 || duration > Duration::from_secs(86400 + 5) {
return Err(Error::new(
"INVALID_ARGUMENT",
"request timeouts must be between one millisecond and one day",
));
}
}
if self.is_closed() {
return Err(Error::new("HOST_DISCONNECTED", "host connection is closed"));
}
let permit = self
.0
.permits
.clone()
.try_acquire_owned()
.map_err(|_| Error::new("QUEUE_FULL", "host request capacity is exhausted"))?;
let (reply, result) = oneshot::channel();
let mut guard = CancelGuard {
state: state.clone(),
retire: abandon_open.then(|| self.0.stop.clone()),
disarmed: false,
};
self.0
.tx
.send(Control::Request(Request {
method: method.to_owned(),
params,
request_timeout,
watchdog,
mutates,
abandon_open,
state,
reply,
permit,
}))
.await
.map_err(|_| Error::new("HOST_DISCONNECTED", "host connection is closed"))?;
let result = result.await.unwrap_or_else(|_| {
Err(Error::new(
"HOST_DISCONNECTED",
"host connection stopped before replying",
))
});
guard.disarmed = true;
result
}
pub async fn shutdown(&self) -> Result<()> {
if !self.is_closed() {
let _ = self.0.tx.send(Control::Shutdown).await;
}
let mut termination = self.0.termination.clone();
loop {
if let Some(result) = termination.borrow().clone() {
return result;
}
termination.changed().await.map_err(|_| {
Error::new(
"HOST_SHUTDOWN_FAILED",
"host cleanup task stopped before confirming process exit",
)
})?;
}
}
}
fn validate_capabilities(value: &Value, mode: HostMode) -> Result<BTreeSet<String>> {
if value["protocolVersion"] != policy()["protocolVersion"] {
return Err(Error::new(
"HOST_HANDSHAKE_FAILED",
"host protocol version is incompatible",
));
}
let values = value["features"]
.as_array()
.ok_or_else(|| Error::new("HOST_HANDSHAKE_FAILED", "host capabilities were malformed"))?;
let features: Option<BTreeSet<_>> = values
.iter()
.map(|value| value.as_str().map(str::to_owned))
.collect();
let features = features
.ok_or_else(|| Error::new("HOST_HANDSHAKE_FAILED", "host features must be strings"))?;
let required = if mode == HostMode::Cli {
"cliRequiredFeatures"
} else {
"documentRequiredFeatures"
};
let expected = policy()[required]
.as_array()
.ok_or_else(|| Error::new("HOST_PROTOCOL_ERROR", "generated host policy is missing"))?;
if expected
.iter()
.any(|feature| !features.contains(feature.as_str().unwrap_or("")))
{
return Err(Error::new(
"HOST_HANDSHAKE_FAILED",
"host lacks a required SDK capability",
));
}
Ok(features)
}
async fn write_frames(
mut stdin: ChildStdin,
mut commands: mpsc::Receiver<Write>,
events: mpsc::Sender<Event>,
) {
while let Some(command) = commands.recv().await {
if command
.state
.phase
.compare_exchange(0, 1, Ordering::AcqRel, Ordering::Acquire)
.is_err()
{
if events.send(Event::Skipped(command.id)).await.is_err() {
break;
}
continue;
}
if stdin.write_all(&command.frame).await.is_err() {
let _ = events
.send(Event::Failed(Error::new(
"HOST_DISCONNECTED",
"could not write a complete host request",
)))
.await;
break;
}
command.state.written.store(true, Ordering::Release);
}
}
#[allow(clippy::too_many_arguments)]
async fn run_actor(
mut child: Child,
mut commands: mpsc::Receiver<Control>,
writes: mpsc::Sender<Write>,
mut events: mpsc::Receiver<Event>,
stop: Arc<Notify>,
closed: Arc<AtomicBool>,
notifications: broadcast::Sender<Value>,
directory: PathBuf,
owned: bool,
shutdown_timeout: Duration,
reader: tokio::task::JoinHandle<()>,
writer: tokio::task::JoinHandle<()>,
stderr: tokio::task::JoinHandle<()>,
termination: watch::Sender<Option<Result<()>>>,
sessions: Sessions,
) {
let mut pending: HashMap<u64, Pending> = HashMap::new();
let mut next_id = 1u64;
let mut shutdown_id = None;
let mut shutdown_deadline = None;
let mut failure = Error::new("HOST_DISCONNECTED", "host connection stopped");
loop {
let next_deadline = pending
.values()
.map(|request| request.deadline)
.chain(shutdown_deadline)
.min()
.unwrap_or_else(|| Instant::now() + Duration::from_secs(86400));
tokio::select! {
_=stop.notified()=>break,
_=child.wait()=>break,
_=tokio::time::sleep_until(next_deadline)=>{
if shutdown_deadline.is_some_and(|deadline|deadline<=Instant::now()){failure=Error::new("TIMEOUT","host shutdown deadline expired");break;}
let expired:Vec<_>=pending.iter().filter_map(|(id,request)|(request.deadline<=Instant::now()).then_some(*id)).collect();
let mut sent_timeout=false;
for id in expired {
let request=pending.get(&id).expect("pending deadline");
if request.state.phase.compare_exchange(0,2,Ordering::AcqRel,Ordering::Acquire).is_err()
&& request.state.phase.load(Ordering::Acquire)==1 {sent_timeout=true;break;}
if let Some(request)=pending.remove(&id) {let _=request.reply.send(Err(Error::new("TIMEOUT","request expired before transmission")));}
}
if sent_timeout {failure=Error::new("TIMEOUT","host response watchdog expired");break;}
}
control=commands.recv(),if shutdown_id.is_none()=>{
match control {
None=>break,
Some(Control::Request(request))=>{
if request.state.phase.load(Ordering::Acquire)==2 {continue;}
let id=next_id;next_id+=1;
let mut frame=json!({"jsonrpc":"2.0","id":id,"method":request.method,"params":request.params});
if let Some(value)=request.request_timeout {frame["requestTimeoutMs"]=json!(value.as_millis() as u64);}
let mut bytes=serde_json::to_vec(&frame).expect("serializable JSON value");bytes.push(b'\n');
if bytes.len()>64*1024*1024 {let _=request.reply.send(Err(Error::new("PAYLOAD_TOO_LARGE","host request exceeded the byte limit")));continue;}
let state=request.state.clone();
pending.insert(id,Pending{reply:request.reply,state:request.state,deadline:Instant::now()+request.watchdog,mutates:request.mutates,abandon_open:request.abandon_open,_permit:request.permit});
match writes.try_send(Write{id,frame:bytes,state}) {
Ok(())=>{},
Err(mpsc::error::TrySendError::Full(_))=>{
if let Some(request)=pending.remove(&id) {let _=request.state.phase.compare_exchange(0,2,Ordering::AcqRel,Ordering::Acquire);let _=request.reply.send(Err(Error::new("QUEUE_FULL","host writer capacity is exhausted")));}
}
Err(mpsc::error::TrySendError::Closed(_))=>{failure=Error::new("HOST_DISCONNECTED","host writer stopped");break;}
}
}
Some(Control::Shutdown)=>{
closed.store(true,Ordering::Release);
let id=next_id;next_id+=1;shutdown_id=Some(id);shutdown_deadline=Some(Instant::now()+shutdown_timeout);
let mut frame=serde_json::to_vec(&json!({"jsonrpc":"2.0","id":id,"method":"host.shutdown","params":{}})).expect("shutdown frame");frame.push(b'\n');
if writes.try_send(Write{id,frame,state:Arc::new(CallState::default())}).is_err(){break;}
}
}
}
event=events.recv()=>{
match event {
None=>break,
Some(Event::Failed(error))=>{failure=error;break;},
Some(Event::Skipped(id))=>{pending.remove(&id);},
Some(Event::Frame(frame))=>{
if frame["jsonrpc"]!="2.0" {failure=Error::new("HOST_PROTOCOL_ERROR","host sent an invalid JSON-RPC version");break;}
if frame.get("id").is_none() && frame["method"].is_string(){
if frame["method"]=="event.sessionClosed" {
if let Some(id)=frame["params"]["sessionId"].as_str(){if let Some(state)=sessions.lock().expect("SDK session lock").get(id){state.store(2,Ordering::Release);}}
}
let _=notifications.send(frame);continue;
}
let Some(id)=frame["id"].as_u64() else {failure=Error::new("HOST_PROTOCOL_ERROR","host response ID was invalid");break;};
if Some(id)==shutdown_id {break;}
let Some(request)=pending.remove(&id) else {failure=Error::new("HOST_PROTOCOL_ERROR","host returned an unknown or duplicate response ID");break;};
let result=decode_response(frame);
if request.abandon_open {
if let Ok(value)=&result {
if let Some(id)=value["data"]["contextId"].as_str().or_else(||value["sessionId"].as_str()) {
let mut sessions=sessions.lock().expect("SDK session lock");
if sessions.get(id).is_none_or(|state|state.load(Ordering::Acquire)>=2){sessions.insert(id.to_owned(),Arc::new(AtomicU8::new(0)));}
}
}
}
if request.abandon_open && request.reply.is_closed() && request.state.phase.load(Ordering::Acquire)==1 {
failure=Error::new("HOST_DISCONNECTED","a cancelled open retired the host to avoid leaking a session");break;
}
let _=request.reply.send(result.map_err(|mut error|{error.outcome_unknown=request.mutates && ["TIMEOUT","HOST_PROTOCOL_ERROR"].contains(&error.code.as_str());error}));
}
}
}
}
}
closed.store(true, Ordering::Release);
for state in sessions.lock().expect("SDK session lock").values() {
if state.load(Ordering::Acquire) < 2 {
state.store(3, Ordering::Release);
}
}
drop(writes);
if shutdown_id.is_none() || failure.code == "TIMEOUT" {
let _ = child.start_kill();
}
let reaped = match timeout(shutdown_timeout, child.wait()).await {
Ok(Ok(_)) => true,
_ => {
let _ = child.start_kill();
matches!(timeout(shutdown_timeout, child.wait()).await, Ok(Ok(_)))
}
};
reader.abort();
writer.abort();
stderr.abort();
let _ = reader.await;
let _ = writer.await;
let _ = stderr.await;
for (_, request) in pending {
if !reaped {
if let Some(asset) = request.state.asset.lock().expect("SDK asset lock").take() {
let _ = asset.keep();
}
}
let mut error = failure.clone();
error.outcome_unknown = request.mutates && request.state.phase.load(Ordering::Acquire) == 1;
let _ = request.reply.send(Err(error));
}
while let Ok(control) = commands.try_recv() {
match control {
Control::Request(request) => {
let _ = request.reply.send(Err(failure.clone()));
}
Control::Shutdown => {}
}
}
if owned && reaped {
let _ = std::fs::remove_dir_all(&directory);
}
termination.send_replace(Some(if reaped {
Ok(())
} else {
Err(Error::new(
"HOST_SHUTDOWN_FAILED",
"host process exit could not be confirmed; state was retained",
))
}));
}
fn decode_response(frame: Value) -> Result<Value> {
if frame.get("result").is_some() == frame.get("error").is_some() {
return Err(Error::new(
"HOST_PROTOCOL_ERROR",
"host response must contain exactly one result or error",
));
}
if let Some(error) = frame.get("error") {
let code = error["code"]
.as_i64()
.ok_or_else(|| Error::new("HOST_PROTOCOL_ERROR", "host error code was invalid"))?;
if !error["message"].is_string() {
return Err(Error::new(
"HOST_PROTOCOL_ERROR",
"host error message was invalid",
));
}
let data = &error["data"];
let sdk_code = if code == -32011 {
"TIMEOUT"
} else {
data["cliCode"]
.as_str()
.or_else(|| data["domainCode"].as_str())
.unwrap_or("HOST_ERROR")
};
let mut result = Error::new(sdk_code, "SuperDoc host rejected the operation");
result.rpc_code = Some(code);
result.details = data.get("details").cloned().map(Box::new);
result.host_message = error["message"].as_str().map(Into::into);
Err(result)
} else {
Ok(frame["result"].clone())
}
}
#[cfg(test)]
mod tests {
use super::*;
fn host(mode: &str, log: &Path) -> HostOptions {
let mut options = HostOptions::new(
std::env::var("SUPERDOC_SDK_TEST_NODE").unwrap_or_else(|_| "node".to_owned()),
);
let file = std::env::var_os("SUPERDOC_SDK_TEST_HOST")
.map(PathBuf::from)
.unwrap_or_else(|| {
PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/host.mjs")
});
options.prefix_args.push(file.to_str().unwrap().to_owned());
options
.env
.insert("SUPERDOC_RUST_TEST_MODE".to_owned(), mode.to_owned());
options.env.insert(
"SUPERDOC_RUST_TEST_LOG".to_owned(),
log.to_str().unwrap().to_owned(),
);
options.shutdown_timeout = Duration::from_millis(250);
options
}
async fn wait_for(mut condition: impl FnMut() -> bool) {
timeout(Duration::from_secs(5), async {
while !condition() {
tokio::time::sleep(Duration::from_millis(5)).await;
}
})
.await
.expect("controlled host barrier was not reached");
}
#[tokio::test]
async fn correlates_out_of_order_replies() {
let directory = tempfile::tempdir().unwrap();
let transport = Transport::connect(host("reorder", &directory.path().join("requests")))
.await
.unwrap();
let a = transport.request(
"cli.invoke",
json!({"argv":["first"]}),
None,
Duration::from_secs(5),
false,
false,
);
let b = transport.request(
"cli.invoke",
json!({"argv":["second"]}),
None,
Duration::from_secs(5),
false,
false,
);
let (a, b) = tokio::join!(a, b);
assert_eq!(a.unwrap()["data"]["marker"], "first");
assert_eq!(b.unwrap()["data"]["marker"], "second");
transport.shutdown().await.unwrap();
}
#[tokio::test]
async fn watchdog_and_disconnection_preserve_uncertain_sent_mutation() {
for mode in ["hang", "exit", "bad-frame", "unknown-id"] {
let directory = tempfile::tempdir().unwrap();
let log = directory.path().join("requests");
let transport = Transport::connect(host(mode, &log)).await.unwrap();
let error = transport
.request(
"cli.invoke",
json!({"argv":["mutation"]}),
None,
Duration::from_secs(1),
true,
false,
)
.await
.unwrap_err();
assert!(
error.outcome_unknown,
"{mode} must not claim a sent mutation was undone"
);
assert!(matches!(
error.code.as_str(),
"TIMEOUT" | "HOST_DISCONNECTED" | "HOST_PROTOCOL_ERROR"
));
transport.shutdown().await.unwrap();
assert_eq!(
std::fs::read_to_string(log)
.unwrap()
.matches("mutation")
.count(),
1,
"a sent mutation must never be replayed"
);
}
}
#[tokio::test]
async fn cancelled_sent_open_retires_host() {
let directory = tempfile::tempdir().unwrap();
let log = directory.path().join("requests");
let transport = Transport::connect(host("hang", &log)).await.unwrap();
let state_directory = transport.state_directory().to_path_buf();
let copy = transport.clone();
let task = tokio::spawn(async move {
copy.request(
"cli.invoke",
json!({"argv":["open"]}),
None,
Duration::from_secs(30),
true,
true,
)
.await
});
wait_for(|| std::fs::read_to_string(&log).is_ok_and(|log| log.contains("cli.invoke")))
.await;
task.abort();
let _ = task.await;
transport.shutdown().await.unwrap();
assert!(!state_directory.exists());
}
#[tokio::test]
async fn queued_cancellation_sends_no_frame_and_releases_capacity() {
let directory = tempfile::tempdir().unwrap();
let log = directory.path().join("requests");
let paused = directory.path().join("paused");
let resume = directory.path().join("resume");
let mut options = host("backpressure", &log);
options.max_pending = 2;
options.env.insert(
"SUPERDOC_RUST_TEST_PAUSED".to_owned(),
paused.to_str().unwrap().to_owned(),
);
options.env.insert(
"SUPERDOC_RUST_TEST_RESUME".to_owned(),
resume.to_str().unwrap().to_owned(),
);
let transport = Transport::connect(options).await.unwrap();
wait_for(|| paused.exists()).await;
let first_state = Arc::new(CallState::default());
let (reply, result) = oneshot::channel();
let permit = transport.0.permits.clone().try_acquire_owned().unwrap();
transport
.0
.tx
.send(Control::Request(Request {
method: "cli.invoke".to_owned(),
params: json!({"argv":["large"],"padding":"x".repeat(16*1024*1024)}),
request_timeout: None,
watchdog: Duration::from_secs(20),
mutates: false,
abandon_open: false,
state: first_state.clone(),
reply,
permit,
}))
.await
.unwrap_or_else(|_| panic!("host command channel closed"));
wait_for(|| first_state.phase.load(Ordering::Acquire) == 1).await;
assert!(
!first_state.written.load(Ordering::Acquire),
"fixture must hold the writer inside the large frame"
);
let copy = transport.clone();
let cancelled = tokio::spawn(async move {
copy.request(
"cli.invoke",
json!({"argv":["must-not-send"]}),
None,
Duration::from_secs(10),
false,
false,
)
.await
});
wait_for(|| transport.0.permits.available_permits() == 0).await;
cancelled.abort();
let _ = cancelled.await;
std::fs::write(resume, b"resume").unwrap();
result.await.unwrap().unwrap();
wait_for(|| transport.0.permits.available_permits() == 2).await;
transport.shutdown().await.unwrap();
assert!(!std::fs::read_to_string(log)
.unwrap()
.contains("must-not-send"));
}
#[test]
fn malformed_rpc_envelopes_do_not_become_success() {
for value in [
json!({"result":{},"error":{}}),
json!({}),
json!({"error":{"code":"wrong","message":"bad"}}),
] {
assert_eq!(
decode_response(value).unwrap_err().code,
"HOST_PROTOCOL_ERROR"
);
}
}
}