use super::*;
type TestResult = Result<(), Box<dyn std::error::Error>>;
#[test]
fn sdk_worker_refusal_is_classified_by_reason_and_preserves_words() -> TestResult {
for code in liminal::protocol::WorkerRefusalCode::ALL {
let words = " busy\nworker_private_words é ";
let error = SdkError::conversation_refusal(9, code.reason_code(), Some(words.to_owned()));
let SdkError::WorkerRefused {
code: name,
words: received,
} = &error
else {
return Err("worker reason was not typed".into());
};
assert_eq!(name, code.name());
assert_eq!(received, words);
assert_eq!(error.to_string(), words);
assert!(!format!("{error:?}").contains("worker_private_words"));
}
let route = SdkError::conversation_refusal(9, 0xffff, Some("worker_busy".to_owned()));
assert!(matches!(route, SdkError::Conversation { .. }));
assert!(matches!(
SdkError::conversation_refusal(9, 0x0401, None),
SdkError::Protocol { .. }
));
Ok(())
}
#[cfg(feature = "embedded")]
#[test]
fn sdk_worker_refuse_reaches_a_real_tcp_caller_and_rejects_oversized_words() -> TestResult {
use crate::{ConnectionPoolConfig, RemoteConfig, RemoteConversationHandle};
use liminal::protocol::WorkerRegistration;
use liminal_server::config::{LimitsConfig, ServerConfig, ServicesConfig};
use liminal_server::server::{connection::ConnectionSupervisor, listener::ServerListener};
let store = TemporaryStore::new()?;
let config = ServerConfig {
listen_address: "127.0.0.1:0".parse()?,
extra_listen_addresses: vec![],
health_listen_address: "127.0.0.1:0".parse()?,
channels: vec![],
routing_rules: vec![],
persistence_path: Some(store.path.clone()),
cluster: None,
auth: None,
drain_timeout_ms: 30_000,
services: ServicesConfig::default(),
limits: LimitsConfig::default(),
participant: None,
websocket: None,
};
let supervisor = ConnectionSupervisor::from_config(&config)?;
let listener = ServerListener::bind(&config, supervisor.clone())?;
let address = listener.local_addr().to_string();
let worker = PushClient::connect_with_registration(
&address,
WorkerRegistration {
namespaces: vec!["refusal".to_owned()],
task_queue: "refusal.request".to_owned(),
node: None,
activity_types: vec!["refusal.request".to_owned()],
identity: "refusal-worker".to_owned(),
activities: vec![],
},
)?;
let failure_wake = Arc::clone(&worker.writer);
let caller = std::thread::spawn(move || -> Result<(), SdkError> {
let result = (|| -> Result<(), SdkError> {
let remote = RemoteConfig::new(
address,
"unused",
"refusal.request",
ConnectionPoolConfig::new(1, 1, 1),
)?
.connect_tcp()?;
let caller = RemoteConversationHandle::new(&remote);
match caller.request(serde_json::json!({"request":"private-input"})) {
Err(SdkError::WorkerRefused { code, words })
if code == "busy" && words == "worker-private-words" =>
{
Ok(())
}
_ => Err(SdkError::Protocol {
description: "real caller did not receive the exact typed refusal".to_owned(),
}),
}
})();
if result.is_err() {
let socket = failure_wake.lock().map_err(|_| SdkError::Connection {
description: "fixture failure-wake lock poisoned".to_owned(),
})?;
socket
.shutdown(Shutdown::Both)
.map_err(|_| SdkError::Connection {
description: "fixture could not signal the caller failure to its worker"
.to_owned(),
})?;
}
result
});
let PushReceiveOutcome::Frame(request) = worker.recv()? else {
caller.join().map_err(|_| "caller panicked")??;
return Err("worker closed before receiving the request".into());
};
let writer = worker.writer_handle();
for error in [
worker.refuse(request.correlation_id(), "busy", &"x".repeat(1025)),
writer.refuse(request.correlation_id(), "busy", &"é".repeat(513)),
] {
let error = error.err().ok_or("oversized refusal was sent")?;
assert!(error.to_string().contains("1024"));
}
assert!(
worker
.refuse(request.correlation_id(), "unknown", "private")
.is_err()
);
writer.refuse(request.correlation_id(), "busy", "worker-private-words")?;
worker.refuse(request.correlation_id(), "internal", "discarded duplicate")?;
worker.reply(
request.correlation_id(),
b"discarded opaque duplicate".to_vec(),
)?;
caller.join().map_err(|_| "caller panicked")??;
drop(writer);
drop(worker);
listener.shutdown()?;
supervisor.shutdown();
drop(supervisor);
store.remove()?;
Ok(())
}
#[cfg(feature = "embedded")]
struct TemporaryStore {
path: std::path::PathBuf,
}
#[cfg(feature = "embedded")]
impl TemporaryStore {
fn new() -> Result<Self, std::io::Error> {
static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
let sequence = NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let path = std::env::temp_dir().join(format!(
"liminal-proto010-{}-{sequence}",
std::process::id()
));
std::fs::create_dir(&path)?;
Ok(Self { path })
}
fn remove(self) -> Result<(), std::io::Error> {
std::fs::remove_dir_all(&self.path)
}
}
#[cfg(feature = "embedded")]
impl Drop for TemporaryStore {
fn drop(&mut self) {
if self.path.exists() {
if let Err(error) = std::fs::remove_dir_all(&self.path) {
eprintln!("owned refusal test store could not be removed: {error}");
}
}
}
}