use std::sync::Arc;
use std::time::{Duration, Instant};
use liminal::conversation::EchoBehaviour;
use liminal::durability::{DurableStore, open_ephemeral};
use liminal::protocol::{
CONVERSATION_REPLY_REQUESTED_FLAG, CausalContext, Frame, MessageEnvelope, SchemaId,
};
use super::apply::apply_frame;
use super::conversation::{ConnectionConversation, ConversationResource};
use super::services::LiminalConnectionServices;
use super::state::{ConnectionProcessState, FrameAction};
use super::supervisor::runtime::ConnectionRuntime;
use crate::ServerError;
use crate::console::records::{canonical_bytes, tap_batch_record};
use crate::console::wall::WallBase;
use crate::tap::{TapFilter, TapSubscription};
type TestResult = Result<(), Box<dyn std::error::Error>>;
#[derive(Debug)]
struct Accepted;
impl ConversationResource for Accepted {
fn message(&self, _: &MessageEnvelope, _: Option<u64>) -> Result<(), ServerError> {
Ok(())
}
fn participant_pids(&self) -> Vec<u64> {
Vec::new()
}
fn has_detected_crash(&self) -> bool {
false
}
fn await_crash(&self, _: Duration) -> Option<Instant> {
None
}
fn receive_reply(&self, _: Duration) -> Result<MessageEnvelope, ServerError> {
Err(ServerError::ListenerAccept {
message: "fixture has no uncorrelated reply".to_owned(),
})
}
fn close(self: Box<Self>) -> Result<(), ServerError> {
Ok(())
}
fn finalize(self: Box<Self>) {}
}
struct Fixture {
runtime: ConnectionRuntime,
state: ConnectionProcessState,
tap: Option<TapSubscription>,
}
impl Fixture {
fn new(served: bool) -> Result<Self, Box<dyn std::error::Error>> {
let config = crate::config::ServerConfig {
listen_address: ([127, 0, 0, 1], 0).into(),
extra_listen_addresses: Vec::new(),
health_listen_address: ([127, 0, 0, 1], 0).into(),
drain_timeout_ms: 30_000,
channels: vec![],
routing_rules: vec![],
persistence_path: None,
cluster: None,
auth: None,
services: crate::config::types::ServicesConfig::default(),
limits: crate::config::types::LimitsConfig::default(),
participant: None,
websocket: None,
};
let store: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
let services = Arc::new(LiminalConnectionServices::from_config_with_store(
&config, store,
)?);
if served {
services.register_responder("jobs.studio", Arc::new(EchoBehaviour))?;
}
services.console().tap().name_station(7, "caller")?;
let tap = services.console().tap().subscribe(TapFilter {
subject: "jobs.studio".to_owned(),
})?;
let runtime = ConnectionRuntime::for_tests(services.clone());
runtime
.signals
.set(services.signals().clone())
.map_err(|_| "signals already installed")?;
let mut state = ConnectionProcessState::default();
state.conversations.insert(
1,
ConnectionConversation::new(Box::new(Accepted), "jobs.studio".to_owned()),
);
Ok(Self {
runtime,
state,
tap: Some(tap),
})
}
fn call(&mut self) -> FrameAction {
apply_frame(
7,
&self.runtime,
&mut self.state,
Frame::ConversationMessage {
flags: CONVERSATION_REPLY_REQUESTED_FLAG,
stream_id: 9,
conversation_id: 1,
envelope: reply(),
},
)
}
fn subscription(&self) -> Result<&TapSubscription, Box<dyn std::error::Error>> {
self.tap.as_ref().ok_or_else(|| "fixture has no tap".into())
}
fn record(&self) -> Result<serde_json::Value, Box<dyn std::error::Error>> {
let batches = self.subscription()?.take_batches()?;
assert_eq!(
batches.len(),
1,
"a service call must reach its subject's tap exactly once"
);
assert_eq!(batches[0].messages.len(), 1);
let wall = WallBase::fixed(1_791_881_244_000, Instant::now());
let record = tap_batch_record("local", "jobs.studio", &batches[0], &wall)?;
Ok(serde_json::from_slice(&canonical_bytes(
&record,
"tap.batch",
)?)?)
}
}
fn reply() -> MessageEnvelope {
MessageEnvelope::new(
SchemaId::new([0; SchemaId::WIRE_LEN]),
CausalContext::independent(),
b"{}".to_vec(),
)
}
#[test]
fn service_tap_answered_correlates_one_request_with_round_trip() -> TestResult {
let mut fixture = Fixture::new(true)?;
let before = fixture.runtime.directory_lookups();
assert!(matches!(fixture.call(), FrameAction::NoResponse));
assert_eq!(
fixture.runtime.directory_lookups() - before,
1,
"the tap must use the actual routing decision, not select a second worker"
);
assert!(
fixture
.state
.pending_replies
.match_reply(1, 999, reply())
.is_none()
);
assert_eq!(
fixture.subscription()?.held()?,
0,
"an unrelated reply settles nothing"
);
assert!(matches!(
fixture.state.pending_replies.match_reply(1, 1, reply()),
Some(Frame::ConversationMessage { stream_id: 9, .. })
));
let value = fixture.record()?;
let message = &value["messages"][0];
assert_eq!(message["subject"], "jobs.studio");
assert_eq!(message["outcome"], "answered");
assert_eq!(message["size"], 2);
assert_eq!(message["from"], "7");
assert_eq!(message["from_station"], "caller");
assert_eq!(message["to_stations"], serde_json::json!(["jobs.studio"]));
assert!(message["fields"]["latency_ms"].is_u64());
assert_eq!(message["hops"][0]["node"], "caller");
assert!(
message["hops"]
.as_array()
.ok_or("hops")?
.iter()
.all(|hop| hop["at_ms"].is_u64())
);
assert!(
message.get("reply_to").is_none(),
"the call is one request record, not a second reply"
);
assert!(
fixture
.state
.pending_replies
.match_reply(1, 1, reply())
.is_none()
);
assert_eq!(fixture.subscription()?.held()?, 0);
let mut late = Fixture::new(true)?;
drop(late.tap.take());
assert!(matches!(late.call(), FrameAction::NoResponse));
late.tap = Some(
late.runtime
.signals
.get()
.and_then(|signals| signals.console())
.ok_or("console")?
.tap()
.subscribe(TapFilter {
subject: "jobs.studio".to_owned(),
})?,
);
assert!(
late.state
.pending_replies
.match_reply(1, 1, reply())
.is_some()
);
assert_eq!(
late.record()?["messages"][0]["outcome"],
"answered",
"a tap joining before the reply still sees the terminal call"
);
assert_round_trip_timing()
}
fn assert_round_trip_timing() -> TestResult {
let tap = crate::tap::Tap::new("local", 64);
let subscription = tap.subscribe(TapFilter {
subject: "jobs.studio".to_owned(),
})?;
let received = Instant::now();
let caller: Arc<str> = Arc::from("caller");
let server: Arc<str> = Arc::from("jobs.studio");
tap.observe_call(&crate::tap::message::CallFact {
subject: "jobs.studio",
size: 2,
connection: 7,
operation: Some(42),
station: Some(&caller),
served_by: Some(&server),
received_at: received,
forwarded_at: received + Duration::from_millis(3),
finished_at: received + Duration::from_millis(17),
outcome: crate::tap::TapOutcome::Answered,
refusal: None,
})?;
let batches = subscription.take_batches()?;
let measured = tap_batch_record(
"local",
"jobs.studio",
&batches[0],
&WallBase::fixed(1_791_881_244_000, received),
)?;
let measured: serde_json::Value =
serde_json::from_slice(&canonical_bytes(&measured, "tap.batch")?)?;
assert_eq!(measured["messages"][0]["fields"]["latency_ms"], 17);
assert_eq!(
measured["messages"][0]["hops"],
serde_json::json!([
{"node": "caller", "at_ms": 0}, {"node": "local", "at_ms": 0},
{"node": "jobs.studio", "at_ms": 3}, {"node": "local", "at_ms": 17},
])
);
Ok(())
}
#[test]
fn service_tap_refused_names_the_server_refusal() -> TestResult {
let mut fixture = Fixture::new(false)?;
let action = fixture.call();
assert!(
matches!(&action, FrameAction::Respond(Frame::ConversationError { message: Some(words), .. }) if words.contains("no_server"))
);
let value = fixture.record()?;
assert_eq!(value["messages"][0]["outcome"], "refused");
assert!(
value["messages"][0]["fields"]["refusal"]
.as_str()
.is_some_and(|words| words.contains("no_server"))
);
let mut capped = Fixture::new(true)?;
capped.state.pending_replies =
super::pending_reply::PendingReplyTable::new(0, 32, Duration::from_secs(5));
assert!(matches!(
capped.call(),
FrameAction::Respond(Frame::ConversationError { .. })
));
let capped_record = capped.record()?;
assert_eq!(capped_record["messages"][0]["outcome"], "refused");
assert_eq!(
capped_record["messages"][0]["to_stations"],
serde_json::json!([]),
"a cap refusal never reached the serving station"
);
Ok(())
}
#[test]
fn service_tap_unanswered_at_deadline_discards_the_late_reply() -> TestResult {
let mut fixture = Fixture::new(true)?;
assert!(matches!(fixture.call(), FrameAction::NoResponse));
assert_eq!(fixture.subscription()?.held()?, 0);
let deadline = Instant::now() + Duration::from_secs(6);
assert_eq!(fixture.state.pending_replies.expire_due(deadline).len(), 1);
let value = fixture.record()?;
assert_eq!(value["messages"][0]["outcome"], "unanswered");
assert!(value["messages"][0]["fields"].get("latency_ms").is_none());
assert!(
fixture
.state
.pending_replies
.match_reply(1, 1, reply())
.is_none()
);
assert!(
fixture
.state
.pending_replies
.expire_due(deadline)
.is_empty()
);
assert_eq!(fixture.subscription()?.held()?, 0);
Ok(())
}