use std::io::{Read, Write};
use std::net::{SocketAddr, TcpStream};
use std::sync::Arc;
use liminal::protocol::{
CONVERSATION_REPLY_REQUESTED_FLAG, CausalContext, Frame, MessageEnvelope, ProtocolError,
ProtocolVersion, SchemaId, WorkerRefusal, WorkerRefusalCode, WorkerRegisterOutcome,
WorkerRegistration, decode, encode, encoded_len,
};
use super::{ConnectionSupervisor, PushReplyDisposition};
use crate::config::{LimitsConfig, ServerConfig, ServicesConfig};
use crate::console::records::{WORKER_REFUSAL_OUTCOME, canonical_bytes, tap_batch_record};
use crate::server::connection::services::LiminalConnectionServices;
use crate::server::listener::ServerListener;
use crate::tap::{TapFilter, TapSubscription};
type TestResult = Result<(), Box<dyn std::error::Error>>;
#[test]
fn worker_refusal_ticket_call_is_error_with_verbatim_words_in_both_eras() -> TestResult {
super::worker_refusal_ticket_fixture::prove()
}
const SUBJECT: &str = "refusal.request";
const STATION: &str = "refusal-worker";
struct Fixture {
listener: ServerListener,
supervisor: ConnectionSupervisor,
services: Arc<LiminalConnectionServices>,
}
impl Fixture {
fn new() -> Result<Self, Box<dyn std::error::Error>> {
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: None,
cluster: None,
auth: None,
drain_timeout_ms: 30_000,
services: ServicesConfig::default(),
limits: LimitsConfig::default(),
participant: None,
websocket: None,
};
let services = Arc::new(LiminalConnectionServices::empty()?);
let supervisor = ConnectionSupervisor::with_services(services.clone())?;
supervisor.install_signals(services.signals().clone())?;
let listener = ServerListener::bind(&config, supervisor.clone())?;
Ok(Self {
listener,
supervisor,
services,
})
}
fn address(&self) -> SocketAddr {
self.listener.local_addr()
}
fn tap(&self) -> Result<TapSubscription, Box<dyn std::error::Error>> {
Ok(self.services.console().tap().subscribe(TapFilter {
subject: SUBJECT.to_owned(),
})?)
}
fn shutdown(self) -> TestResult {
self.listener.shutdown()?;
self.supervisor.shutdown();
self.services.conversation_supervisor().shutdown();
Ok(())
}
}
struct Peer {
stream: TcpStream,
buffer: Vec<u8>,
}
impl Peer {
fn connect(address: SocketAddr) -> Result<Self, Box<dyn std::error::Error>> {
let mut peer = Self {
stream: TcpStream::connect(address)?,
buffer: vec![],
};
peer.stream.set_nodelay(true)?;
peer.send(&Frame::Connect {
flags: 0,
min_version: ProtocolVersion::new(1, 0),
max_version: ProtocolVersion::new(1, 1),
auth_token: vec![],
})?;
assert!(matches!(peer.read()?, Frame::ConnectAck { .. }));
Ok(peer)
}
fn worker(address: SocketAddr) -> Result<Self, Box<dyn std::error::Error>> {
let mut peer = Self::connect(address)?;
peer.send(&Frame::WorkerRegister {
flags: 0,
registration: WorkerRegistration {
namespaces: vec!["refusal".to_owned()],
task_queue: SUBJECT.to_owned(),
node: None,
activity_types: vec![SUBJECT.to_owned()],
identity: STATION.to_owned(),
activities: vec![],
},
})?;
assert!(matches!(
peer.read()?,
Frame::WorkerRegisterAck {
outcome: WorkerRegisterOutcome::Accepted,
..
}
));
Ok(peer)
}
fn send(&mut self, frame: &Frame) -> TestResult {
let mut bytes = vec![0; encoded_len(frame)?];
let count = encode(frame, &mut bytes)?;
assert_eq!(count, bytes.len());
self.stream.write_all(&bytes)?;
Ok(())
}
fn raw_reply(&mut self, correlation_id: u64, flags: u8, payload: Vec<u8>) -> TestResult {
let frame = Frame::new_push_reply(1, correlation_id, payload)?;
let mut bytes = vec![0; encoded_len(&frame)?];
encode(&frame, &mut bytes)?;
*bytes.get_mut(1).ok_or("missing flag byte")? = flags;
self.stream.write_all(&bytes)?;
Ok(())
}
fn read(&mut self) -> Result<Frame, Box<dyn std::error::Error>> {
loop {
match decode(&self.buffer) {
Ok((frame, consumed)) => {
self.buffer.drain(..consumed);
return Ok(frame);
}
Err(
ProtocolError::IncompleteHeader { .. } | ProtocolError::TruncatedPayload { .. },
) => {
let mut chunk = [0; 4096];
let count = self.stream.read(&mut chunk)?;
if count == 0 {
return Err("peer closed before the required frame".into());
}
self.buffer
.extend_from_slice(chunk.get(..count).ok_or("invalid socket count")?);
}
Err(error) => return Err(error.into()),
}
}
}
fn request(&mut self, conversation_id: u64) -> TestResult {
self.send(&Frame::ConversationOpen {
flags: 0,
stream_id: 1,
conversation_id,
subject: SUBJECT.to_owned(),
})?;
self.send(&Frame::ConversationMessage {
flags: CONVERSATION_REPLY_REQUESTED_FLAG,
stream_id: 1,
conversation_id,
envelope: MessageEnvelope::new(
SchemaId::new([0; 32]),
CausalContext::independent(),
b"private-request-body".to_vec(),
),
})
}
fn barrier(&mut self) -> TestResult {
self.send(&Frame::Ping { flags: 0 })?;
assert!(matches!(self.read()?, Frame::Pong { .. }));
Ok(())
}
}
fn push(frame: Frame) -> Result<u64, Box<dyn std::error::Error>> {
let Frame::Push {
correlation_id,
payload,
..
} = frame
else {
return Err("worker received no push".into());
};
assert_eq!(payload, b"private-request-body");
Ok(correlation_id)
}
fn refused(frame: Frame, conversation: u64, code: WorkerRefusalCode, words: &str) -> TestResult {
let Frame::ConversationError {
conversation_id,
stream_id,
reason_code,
message,
..
} = frame
else {
return Err("caller received no named refusal".into());
};
assert_eq!(conversation_id, conversation);
assert_eq!(stream_id, 1);
assert_eq!(reason_code, code.reason_code());
assert_eq!(message.as_deref(), Some(words));
Ok(())
}
#[test]
fn worker_refusal_real_server_preserves_all_codes_words_and_terminal_tap() -> TestResult {
let fixture = Fixture::new()?;
let tap = fixture.tap()?;
let mut worker = Peer::worker(fixture.address())?;
let mut caller = Peer::connect(fixture.address())?;
for (index, code) in WorkerRefusalCode::ALL.into_iter().enumerate() {
let conversation = u64::try_from(index)? + 1;
caller.request(conversation)?;
let correlation = push(worker.read()?)?;
let words = " worker-private-words\n é ";
worker.send(&Frame::new_push_refusal(
1,
correlation,
&WorkerRefusal::new(code.name(), words)?,
)?)?;
refused(caller.read()?, conversation, code, words)?;
let batches = tap.take_batches()?;
assert_eq!(batches.len(), 1);
assert_eq!(batches[0].messages.len(), 1);
let record = tap_batch_record(
"local",
SUBJECT,
&batches[0],
&crate::console::wall::WallBase::fixed(1, std::time::Instant::now()),
)?;
assert_eq!(record.messages[0].outcome, WORKER_REFUSAL_OUTCOME);
assert_eq!(
record.messages[0]
.fields
.as_ref()
.and_then(|fields| fields.refusal.as_deref()),
Some(code.caller_name())
);
let encoded = canonical_bytes(&record, "tap.batch")?;
let encoded = String::from_utf8(encoded)?;
assert!(!encoded.contains("worker-private-words"));
assert!(!encoded.contains("private-request-body"));
worker.send(&Frame::new_push_refusal(
1,
correlation,
&WorkerRefusal::new("internal", "duplicate")?,
)?)?;
worker.send(&Frame::new_push_reply(
1,
correlation,
b"duplicate".to_vec(),
)?)?;
worker.barrier()?;
assert!(tap.take_batches()?.is_empty());
let runtime = fixture.supervisor.runtime_for_test();
let pid = runtime
.records
.lock()
.map_err(|_| "registry poisoned")?
.connections
.iter()
.find_map(|(pid, record)| record.registration.as_ref().map(|_| *pid))
.ok_or("worker not recorded")?;
assert_eq!(runtime.pending_push_count_for(pid), 0);
assert_eq!(
runtime.resolve_push(pid, correlation, vec![]),
PushReplyDisposition::UnknownCorrelation
);
}
caller.request(99)?;
let correlation = push(worker.read()?)?;
worker.send(&Frame::new_push_reply(1, correlation, vec![0, 1, 2, 255])?)?;
let Frame::ConversationMessage {
conversation_id: 99,
envelope,
..
} = caller.read()?
else {
return Err("opaque reply changed".into());
};
assert_eq!(envelope.payload, [0, 1, 2, 255]);
drop(caller);
drop(worker);
drop(tap);
fixture.shutdown()
}
#[test]
fn worker_refusal_real_server_malformed_and_reserved_flags_answer_unavailable() -> TestResult {
let fixture = Fixture::new()?;
let mut worker = Peer::worker(fixture.address())?;
let mut caller = Peer::connect(fixture.address())?;
for (index, (flags, payload)) in [
(1, b"private-malformed-body".to_vec()),
(
1,
br#"{"code":"busy","words":"private","extra":true}"#.to_vec(),
),
(2, b"private-reserved-body".to_vec()),
(3, WorkerRefusal::new("busy", "private")?.canonical_bytes()?),
]
.into_iter()
.enumerate()
{
let conversation = u64::try_from(index)? + 1;
caller.request(conversation)?;
let correlation = push(worker.read()?)?;
worker.raw_reply(correlation, flags, payload)?;
refused(
caller.read()?,
conversation,
WorkerRefusalCode::Unavailable,
&format!("worker {STATION} sent a malformed refusal"),
)?;
}
drop(caller);
drop(worker);
fixture.shutdown()
}
#[test]
fn worker_refusal_real_server_cannot_consume_another_workers_slot() -> TestResult {
let fixture = Fixture::new()?;
let mut worker = Peer::worker(fixture.address())?;
let mut attacker = Peer::connect(fixture.address())?;
let mut caller = Peer::connect(fixture.address())?;
caller.request(1)?;
let correlation = push(worker.read()?)?;
let refusal = WorkerRefusal::new("busy", "authentic worker words")?;
attacker.send(&Frame::new_push_refusal(
1,
correlation,
&WorkerRefusal::new("internal", "forged words")?,
)?)?;
attacker.barrier()?;
worker.send(&Frame::new_push_refusal(1, correlation, &refusal)?)?;
refused(caller.read()?, 1, WorkerRefusalCode::Busy, refusal.words())?;
drop(caller);
drop(attacker);
drop(worker);
fixture.shutdown()
}
#[test]
fn worker_refusal_fault_log_names_station_and_never_formats_payload() -> TestResult {
let fixture = Fixture::new()?;
let worker = Peer::worker(fixture.address())?;
let runtime = fixture.supervisor.runtime_for_test();
let pid = runtime
.records
.lock()
.map_err(|_| "registry poisoned")?
.connections
.iter()
.find_map(|(pid, record)| record.registration.as_ref().map(|_| *pid))
.ok_or("worker not recorded")?;
let awaiter = runtime.push_for_tests(pid, Some(SUBJECT))?;
let mut state = crate::server::connection::state::ConnectionProcessState::default();
let log = crate::test_log::CapturedLog::default();
let action = log.capture(|| {
crate::server::connection::apply::apply_frame(
pid,
&runtime,
&mut state,
Frame::PushReply {
flags: 1,
stream_id: 1,
correlation_id: awaiter.correlation_id(),
payload: b"private-malformed-body".to_vec(),
},
)
});
assert!(matches!(
action,
crate::server::connection::state::FrameAction::NoResponse
));
let logged = log.contents();
assert!(logged.contains("worker_refusal_malformed_record"));
assert!(logged.contains(STATION));
assert!(!logged.contains("private-malformed-body"));
assert!(!logged.contains("112, 114, 105"));
assert!(matches!(
awaiter.try_receive(),
Some(Err(super::PushReplyFailure::Worker(_)))
));
drop(worker);
fixture.shutdown()
}
#[test]
fn worker_refusal_station_bound_keeps_the_full_fault_within_the_words_bound() -> TestResult {
let fixture = Fixture::new()?;
let bound = crate::server::connection::apply::WORKER_REFUSAL_STATION_MAX;
for extra in [0, 1] {
let identity = "s".repeat(bound + extra);
let mut worker = Peer::connect(fixture.address())?;
worker.send(&Frame::WorkerRegister {
flags: 0,
registration: WorkerRegistration {
namespaces: vec!["refusal".to_owned()],
task_queue: SUBJECT.to_owned(),
node: None,
activity_types: vec![SUBJECT.to_owned()],
identity: identity.clone(),
activities: vec![],
},
})?;
if extra == 0 {
assert!(matches!(
worker.read()?,
Frame::WorkerRegisterAck {
outcome: WorkerRegisterOutcome::Accepted,
..
}
));
let mut caller = Peer::connect(fixture.address())?;
caller.request(1)?;
let correlation = push(worker.read()?)?;
worker.raw_reply(correlation, 1, b"malformed".to_vec())?;
let words = format!("worker {identity} sent a malformed refusal");
assert_eq!(words.len(), liminal::protocol::WORKER_REFUSAL_WORDS_MAX);
refused(caller.read()?, 1, WorkerRefusalCode::Unavailable, &words)?;
} else {
let Frame::WorkerRegisterAck {
outcome: WorkerRegisterOutcome::Rejected { reason },
..
} = worker.read()?
else {
return Err("oversized station identity was accepted".into());
};
assert!(reason.contains("worker_station_too_long"));
assert!(reason.contains(&bound.to_string()));
}
}
fixture.shutdown()
}