9bed780cb9
pending_connections_are_bounded_and_hello_expires gave the router 20 ms to register a connection and then slept 80 ms past a 40 ms hello timeout before reading the count once. Both are claims about how fast this box is, not about the router: the first failed once in 20 whole-crate runs beside four busy loops. Both now wait for the event they are about, under the suite's ordinary ten-second bound, so the test still fails if a pending connection never registers or a pending Hello never expires.
834 lines
28 KiB
Rust
834 lines
28 KiB
Rust
mod common;
|
|
|
|
use common::{Via, env, env_with, obj, within};
|
|
use flybus::wire::{Envelope, Kind, contract_digest, read_frame, write_frame};
|
|
use flybus::{
|
|
CancelState, Client, ClientConfig, Dispatch, ErrorCode, Limits, Policy, Retained, Router,
|
|
RouterConfig, ServiceConfig, SubscriptionConfig, Transport,
|
|
};
|
|
use serde_json::{Map, Value, json};
|
|
use std::io;
|
|
use std::pin::Pin;
|
|
use std::task::{Context, Poll, Waker};
|
|
use std::time::Duration;
|
|
use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
|
|
use tokio::sync::oneshot;
|
|
|
|
struct WriteFailsAfterHello {
|
|
response: Vec<u8>,
|
|
read: usize,
|
|
writes: usize,
|
|
}
|
|
|
|
impl AsyncRead for WriteFailsAfterHello {
|
|
fn poll_read(
|
|
mut self: Pin<&mut Self>,
|
|
_cx: &mut Context<'_>,
|
|
buf: &mut ReadBuf<'_>,
|
|
) -> Poll<io::Result<()>> {
|
|
if self.read == self.response.len() {
|
|
return Poll::Pending;
|
|
}
|
|
let n = buf.remaining().min(self.response.len() - self.read);
|
|
let end = self.read + n;
|
|
buf.put_slice(&self.response[self.read..end]);
|
|
self.read = end;
|
|
Poll::Ready(Ok(()))
|
|
}
|
|
}
|
|
|
|
impl AsyncWrite for WriteFailsAfterHello {
|
|
fn poll_write(
|
|
mut self: Pin<&mut Self>,
|
|
_cx: &mut Context<'_>,
|
|
buf: &[u8],
|
|
) -> Poll<io::Result<usize>> {
|
|
if self.writes == 0 {
|
|
self.writes = 1;
|
|
Poll::Ready(Ok(buf.len()))
|
|
} else {
|
|
Poll::Ready(Err(io::Error::new(io::ErrorKind::BrokenPipe, "injected")))
|
|
}
|
|
}
|
|
|
|
fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
|
|
Poll::Ready(Ok(()))
|
|
}
|
|
|
|
fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
|
|
Poll::Ready(Ok(()))
|
|
}
|
|
}
|
|
|
|
struct WriteBlocksAsReaderFails {
|
|
response: Vec<u8>,
|
|
read: usize,
|
|
hello_written: bool,
|
|
command_write_started: bool,
|
|
reader_waker: Option<Waker>,
|
|
dropped: Option<oneshot::Sender<()>>,
|
|
}
|
|
|
|
impl Drop for WriteBlocksAsReaderFails {
|
|
fn drop(&mut self) {
|
|
if let Some(dropped) = self.dropped.take() {
|
|
let _ = dropped.send(());
|
|
}
|
|
}
|
|
}
|
|
|
|
impl AsyncRead for WriteBlocksAsReaderFails {
|
|
fn poll_read(
|
|
mut self: Pin<&mut Self>,
|
|
cx: &mut Context<'_>,
|
|
buf: &mut ReadBuf<'_>,
|
|
) -> Poll<io::Result<()>> {
|
|
if self.read < self.response.len() {
|
|
let n = buf.remaining().min(self.response.len() - self.read);
|
|
let end = self.read + n;
|
|
buf.put_slice(&self.response[self.read..end]);
|
|
self.read = end;
|
|
return Poll::Ready(Ok(()));
|
|
}
|
|
if self.command_write_started {
|
|
return Poll::Ready(Err(io::Error::new(
|
|
io::ErrorKind::ConnectionReset,
|
|
"injected reader failure",
|
|
)));
|
|
}
|
|
self.reader_waker = Some(cx.waker().clone());
|
|
Poll::Pending
|
|
}
|
|
}
|
|
|
|
impl AsyncWrite for WriteBlocksAsReaderFails {
|
|
fn poll_write(
|
|
mut self: Pin<&mut Self>,
|
|
_cx: &mut Context<'_>,
|
|
buf: &[u8],
|
|
) -> Poll<io::Result<usize>> {
|
|
if !self.hello_written {
|
|
self.hello_written = true;
|
|
return Poll::Ready(Ok(buf.len()));
|
|
}
|
|
self.command_write_started = true;
|
|
if let Some(waker) = self.reader_waker.take() {
|
|
waker.wake();
|
|
}
|
|
Poll::Pending
|
|
}
|
|
|
|
fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
|
|
Poll::Ready(Ok(()))
|
|
}
|
|
|
|
fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
|
|
Poll::Pending
|
|
}
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn responder_survives_cancel_then_request_drop() {
|
|
let e = env(Via::Memory).await;
|
|
let server = e.client("server").await;
|
|
let caller = e.client("caller").await;
|
|
let mut service = server
|
|
.register("example.race", ServiceConfig::default())
|
|
.await
|
|
.unwrap();
|
|
let pending = caller
|
|
.call("example.race", None, "Work", obj(json!({})), &[])
|
|
.await
|
|
.unwrap();
|
|
let request = within("request", service.next()).await.unwrap();
|
|
let responder = request.responder();
|
|
|
|
assert_eq!(
|
|
pending.cancel().await.unwrap(),
|
|
CancelState::ExecutionUnknown
|
|
);
|
|
drop(request);
|
|
e.settle("request consumed", |s| s.owners == 0).await;
|
|
|
|
let routed = responder.reply(obj(json!({"done": true})), &[]).await;
|
|
assert_eq!(routed, Ok(false), "a detached reply is still a valid reply");
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn cancel_after_request_consumption_retires_correlation() {
|
|
let e = env(Via::Memory).await;
|
|
let server = e.client("server").await;
|
|
let caller = e.client("caller").await;
|
|
let mut service = server
|
|
.register("example.leak", ServiceConfig::default())
|
|
.await
|
|
.unwrap();
|
|
let mut pending = caller
|
|
.call("example.leak", None, "Work", obj(json!({})), &[])
|
|
.await
|
|
.unwrap();
|
|
let request = within("request", service.next()).await.unwrap();
|
|
drop(request);
|
|
let failure = within("last responder terminal failure", pending.result()).await;
|
|
let failure = failure.unwrap_err();
|
|
assert_eq!(
|
|
(failure.code, failure.dispatch),
|
|
(ErrorCode::CallGone, Dispatch::Dispatched)
|
|
);
|
|
e.settle("consumed call retired", |s| {
|
|
s.calls == 0 && s.active_calls == 0 && s.owners == 0 && s.reply_capabilities == 0
|
|
})
|
|
.await;
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn retained_replay_obeys_bounded_queue_byte_quota() {
|
|
let limits = Limits {
|
|
max_queued_bytes_per_client: 1,
|
|
max_owners_per_client: 2,
|
|
reserved_owners_per_client: 1,
|
|
..Limits::default()
|
|
};
|
|
let e = env_with(Via::Memory, limits, Policy::open()).await;
|
|
let publisher = e.client("publisher").await;
|
|
let reader = e.client("reader").await;
|
|
publisher
|
|
.declare_topic("t.replay", Retained::Latest)
|
|
.await
|
|
.unwrap();
|
|
publisher
|
|
.publish("t.replay", obj(json!({"large": "x".repeat(1024)})), &[])
|
|
.await
|
|
.unwrap();
|
|
|
|
// Exhaust the reader's ordinary-owner allowance so replay cannot immediately dispatch.
|
|
let _hold = reader
|
|
.artifacts()
|
|
.allocate(0, "application/octet-stream")
|
|
.await
|
|
.unwrap()
|
|
.seal()
|
|
.await
|
|
.unwrap();
|
|
let replay = reader
|
|
.subscribe("t.replay", SubscriptionConfig::bounded().replay(true))
|
|
.await;
|
|
assert_eq!(replay.unwrap_err().code, ErrorCode::Backpressure);
|
|
assert_eq!(e.stats().subscriptions, 0, "failed replay is atomic");
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn latest_replay_remains_bounded_outside_the_bounded_byte_pool() {
|
|
let limits = Limits {
|
|
max_queued_bytes_per_client: 1,
|
|
max_owners_per_client: 2,
|
|
reserved_owners_per_client: 1,
|
|
..Limits::default()
|
|
};
|
|
let e = env_with(Via::Memory, limits, Policy::open()).await;
|
|
let publisher = e.client("publisher").await;
|
|
let reader = e.client("reader").await;
|
|
publisher
|
|
.declare_topic("t.latest", Retained::Latest)
|
|
.await
|
|
.unwrap();
|
|
publisher
|
|
.publish("t.latest", obj(json!({"large": "x".repeat(1024)})), &[])
|
|
.await
|
|
.unwrap();
|
|
let _hold = reader
|
|
.artifacts()
|
|
.allocate(0, "application/octet-stream")
|
|
.await
|
|
.unwrap()
|
|
.seal()
|
|
.await
|
|
.unwrap();
|
|
let sub = reader
|
|
.subscribe("t.latest", SubscriptionConfig::latest().replay(true))
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(e.stats().subscriptions, 1);
|
|
drop(sub);
|
|
e.settle("latest replay released", |s| s.subscriptions == 0)
|
|
.await;
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn pending_connections_are_bounded_and_hello_expires() {
|
|
let dir = tempfile::tempdir().unwrap();
|
|
let mut config = RouterConfig::new(dir.path());
|
|
config.limits.max_clients = 1;
|
|
config.hello_timeout = Duration::from_millis(40);
|
|
config.policy = Policy::closed()
|
|
.client("first", flybus::Grants::all())
|
|
.client("second", flybus::Grants::all());
|
|
let router = Router::new(config).unwrap();
|
|
|
|
let pending = router.connect_in_memory_as("first");
|
|
// Registration happens on the router's own task. How long that takes is this box's
|
|
// business; that it happens is the router's.
|
|
within("the pending connection occupies the only slot", async {
|
|
while router.stats().connections != 1 {
|
|
tokio::time::sleep(Duration::from_millis(1)).await;
|
|
}
|
|
})
|
|
.await;
|
|
assert_eq!(router.stats().connections, 1);
|
|
let refused = Client::connect(
|
|
router.connect_in_memory_as("second"),
|
|
ClientConfig::new("second", dir.path()),
|
|
)
|
|
.await;
|
|
assert_eq!(refused.unwrap_err().code, ErrorCode::RouterLost);
|
|
// The 40 ms hello timeout expires on the router's clock: wait for the expiry to be
|
|
// observed rather than sleep past it and read the count once.
|
|
within("the pending Hello expires", async {
|
|
while router.stats().connections != 0 {
|
|
tokio::time::sleep(Duration::from_millis(1)).await;
|
|
}
|
|
})
|
|
.await;
|
|
assert_eq!(router.stats().connections, 0, "pending Hello timed out");
|
|
drop(pending);
|
|
|
|
let second = Client::connect(
|
|
router.connect_in_memory_as("second"),
|
|
ClientConfig::new("second", dir.path()),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
second.close().await;
|
|
let unbound = Client::connect(
|
|
router.connect_in_memory(),
|
|
ClientConfig::new("first", dir.path()),
|
|
)
|
|
.await;
|
|
assert_eq!(unbound.unwrap_err().code, ErrorCode::RouterLost);
|
|
}
|
|
|
|
fn reply_body(value: Value) -> Map<String, Value> {
|
|
obj(json!({"ok": true, "value": value}))
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn client_rejects_wrong_version_and_operation_from_router() {
|
|
let dir = tempfile::tempdir().unwrap();
|
|
let (client_stream, mut fake_router) = tokio::io::duplex(64 * 1024);
|
|
let server = tokio::spawn(async move {
|
|
let hello = read_frame(&mut fake_router).await.unwrap().unwrap();
|
|
let hello = Envelope::decode(&hello).unwrap();
|
|
let hello_reply = Envelope {
|
|
major: 1,
|
|
minor: 0,
|
|
id: "bus-1".into(),
|
|
reply_to: Some(hello.id),
|
|
kind: Kind::Reply,
|
|
op: "bus.hello".into(),
|
|
body: reply_body(json!({
|
|
"routerId": "router-fake",
|
|
"connectionId": "conn-1",
|
|
"selectedMajor": 1,
|
|
"selectedMinor": 0,
|
|
"contractDigest": contract_digest(),
|
|
"limits": Limits::default().to_json(),
|
|
})),
|
|
attachments: Vec::new(),
|
|
};
|
|
write_frame(&mut fake_router, &hello_reply.encode().unwrap())
|
|
.await
|
|
.unwrap();
|
|
|
|
let command = read_frame(&mut fake_router).await.unwrap().unwrap();
|
|
let command = Envelope::decode(&command).unwrap();
|
|
let invalid_reply = Envelope {
|
|
major: 2,
|
|
minor: 0,
|
|
id: "bus-1".into(),
|
|
reply_to: Some(command.id),
|
|
kind: Kind::Reply,
|
|
op: "publish".into(),
|
|
body: reply_body(json!({
|
|
"declared": true,
|
|
"topicIncarnation": "top-1",
|
|
})),
|
|
attachments: Vec::new(),
|
|
};
|
|
write_frame(&mut fake_router, &invalid_reply.encode().unwrap())
|
|
.await
|
|
.unwrap();
|
|
});
|
|
|
|
let client = Client::connect(
|
|
Transport::from_stream(client_stream),
|
|
ClientConfig::new("client", dir.path()),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
let result = client.declare_topic("t.x", Retained::None).await;
|
|
assert!(
|
|
result.is_err(),
|
|
"invalid router envelope was accepted: {result:?}"
|
|
);
|
|
server.await.unwrap();
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn client_strictly_validates_hello_envelope() {
|
|
for case in ["version", "operation", "id", "attachments"] {
|
|
let dir = tempfile::tempdir().unwrap();
|
|
let (client_stream, mut fake_router) = tokio::io::duplex(64 * 1024);
|
|
let server = tokio::spawn(async move {
|
|
let hello =
|
|
Envelope::decode(&read_frame(&mut fake_router).await.unwrap().unwrap()).unwrap();
|
|
let mut reply = Envelope {
|
|
major: 1,
|
|
minor: 0,
|
|
id: "bus-1".into(),
|
|
reply_to: Some(hello.id),
|
|
kind: Kind::Reply,
|
|
op: "bus.hello".into(),
|
|
body: reply_body(json!({
|
|
"routerId": "router-fake",
|
|
"connectionId": "conn-1",
|
|
"selectedMajor": 1,
|
|
"selectedMinor": 0,
|
|
"contractDigest": contract_digest(),
|
|
"limits": Limits::default().to_json(),
|
|
})),
|
|
attachments: Vec::new(),
|
|
};
|
|
match case {
|
|
"version" => reply.major = 2,
|
|
"operation" => reply.op = "publish".into(),
|
|
"id" => reply.id = "msg-1".into(),
|
|
"attachments" => reply.attachments.push(flybus::wire::Attachment {
|
|
name: "x".into(),
|
|
reference: flybus::ArtifactRef {
|
|
store_id: "store-fake".into(),
|
|
artifact_id: "a-1".into(),
|
|
generation: 1,
|
|
byte_length: 0,
|
|
content_type: "x/y".into(),
|
|
digest: None,
|
|
},
|
|
owner_id: "own-1".into(),
|
|
}),
|
|
_ => unreachable!(),
|
|
}
|
|
write_frame(&mut fake_router, &reply.encode().unwrap())
|
|
.await
|
|
.unwrap();
|
|
});
|
|
let result = Client::connect(
|
|
Transport::from_stream(client_stream),
|
|
ClientConfig::new("client", dir.path()),
|
|
)
|
|
.await;
|
|
assert!(result.is_err(), "invalid Hello case {case} was accepted");
|
|
server.await.unwrap();
|
|
}
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn client_rejects_unknown_reply_fields_and_invalid_direction_rules() {
|
|
let dir = tempfile::tempdir().unwrap();
|
|
let (client_stream, mut fake_router) = tokio::io::duplex(64 * 1024);
|
|
let server = tokio::spawn(async move {
|
|
let hello =
|
|
Envelope::decode(&read_frame(&mut fake_router).await.unwrap().unwrap()).unwrap();
|
|
let hello_reply = Envelope {
|
|
major: 1,
|
|
minor: 0,
|
|
id: "bus-1".into(),
|
|
reply_to: Some(hello.id),
|
|
kind: Kind::Reply,
|
|
op: "bus.hello".into(),
|
|
body: reply_body(json!({
|
|
"routerId": "router-fake",
|
|
"connectionId": "conn-1",
|
|
"selectedMajor": 1,
|
|
"selectedMinor": 0,
|
|
"contractDigest": contract_digest(),
|
|
"limits": Limits::default().to_json(),
|
|
})),
|
|
attachments: Vec::new(),
|
|
};
|
|
write_frame(&mut fake_router, &hello_reply.encode().unwrap())
|
|
.await
|
|
.unwrap();
|
|
let command =
|
|
Envelope::decode(&read_frame(&mut fake_router).await.unwrap().unwrap()).unwrap();
|
|
let invalid = Envelope {
|
|
major: 1,
|
|
minor: 0,
|
|
id: "bus-2".into(),
|
|
reply_to: Some(command.id),
|
|
kind: Kind::Reply,
|
|
op: "topic.declare".into(),
|
|
body: reply_body(json!({
|
|
"declared": true,
|
|
"topicIncarnation": "top-1",
|
|
"extra": true,
|
|
})),
|
|
attachments: Vec::new(),
|
|
};
|
|
write_frame(&mut fake_router, &invalid.encode().unwrap())
|
|
.await
|
|
.unwrap();
|
|
});
|
|
let client = Client::connect(
|
|
Transport::from_stream(client_stream),
|
|
ClientConfig::new("client", dir.path()),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
assert!(client.declare_topic("t.x", Retained::None).await.is_err());
|
|
server.await.unwrap();
|
|
|
|
let (client_stream, mut fake_router) = tokio::io::duplex(64 * 1024);
|
|
let server = tokio::spawn(async move {
|
|
let hello =
|
|
Envelope::decode(&read_frame(&mut fake_router).await.unwrap().unwrap()).unwrap();
|
|
let hello_reply = Envelope {
|
|
major: 1,
|
|
minor: 0,
|
|
id: "bus-1".into(),
|
|
reply_to: Some(hello.id),
|
|
kind: Kind::Reply,
|
|
op: "bus.hello".into(),
|
|
body: reply_body(json!({
|
|
"routerId": "router-fake",
|
|
"connectionId": "conn-1",
|
|
"selectedMajor": 1,
|
|
"selectedMinor": 0,
|
|
"contractDigest": contract_digest(),
|
|
"limits": Limits::default().to_json(),
|
|
})),
|
|
attachments: Vec::new(),
|
|
};
|
|
write_frame(&mut fake_router, &hello_reply.encode().unwrap())
|
|
.await
|
|
.unwrap();
|
|
let invalid_notice = Envelope {
|
|
major: 1,
|
|
minor: 0,
|
|
id: "bus-2".into(),
|
|
reply_to: Some("msg-99".into()),
|
|
kind: Kind::Notice,
|
|
op: "connection.closing".into(),
|
|
body: obj(json!({"code": "ROUTER_LOST", "message": "bad direction"})),
|
|
attachments: Vec::new(),
|
|
};
|
|
write_frame(&mut fake_router, &invalid_notice.encode().unwrap())
|
|
.await
|
|
.unwrap();
|
|
});
|
|
let client = Client::connect(
|
|
Transport::from_stream(client_stream),
|
|
ClientConfig::new("client", dir.path()),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
tokio::time::timeout(Duration::from_secs(1), async {
|
|
while client.closed().is_none() {
|
|
tokio::task::yield_now().await;
|
|
}
|
|
})
|
|
.await
|
|
.unwrap();
|
|
server.await.unwrap();
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn writer_failure_terminates_reader_and_pending_work() {
|
|
let dir = tempfile::tempdir().unwrap();
|
|
let hello_reply = Envelope {
|
|
major: 1,
|
|
minor: 0,
|
|
id: "bus-1".into(),
|
|
reply_to: Some("msg-1".into()),
|
|
kind: Kind::Reply,
|
|
op: "bus.hello".into(),
|
|
body: reply_body(json!({
|
|
"routerId": "router-fake",
|
|
"connectionId": "conn-1",
|
|
"selectedMajor": 1,
|
|
"selectedMinor": 0,
|
|
"contractDigest": contract_digest(),
|
|
"limits": Limits::default().to_json(),
|
|
})),
|
|
attachments: Vec::new(),
|
|
}
|
|
.encode()
|
|
.unwrap();
|
|
let mut response = (hello_reply.len() as u32).to_le_bytes().to_vec();
|
|
response.extend(hello_reply);
|
|
let client = Client::connect(
|
|
Transport::from_stream(WriteFailsAfterHello {
|
|
response,
|
|
read: 0,
|
|
writes: 0,
|
|
}),
|
|
ClientConfig::new("client", dir.path()),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
let result = tokio::time::timeout(
|
|
Duration::from_secs(1),
|
|
client.declare_topic("t.x", Retained::None),
|
|
)
|
|
.await
|
|
.expect("writer failure left command pending");
|
|
assert_eq!(result.unwrap_err().code, ErrorCode::RouterLost);
|
|
tokio::time::timeout(Duration::from_secs(1), client.close())
|
|
.await
|
|
.expect("writer failure left close waiting");
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn reader_failure_cancels_blocked_write_and_drops_transport() {
|
|
let dir = tempfile::tempdir().unwrap();
|
|
let hello_reply = Envelope {
|
|
major: 1,
|
|
minor: 0,
|
|
id: "bus-1".into(),
|
|
reply_to: Some("msg-1".into()),
|
|
kind: Kind::Reply,
|
|
op: "bus.hello".into(),
|
|
body: reply_body(json!({
|
|
"routerId": "router-fake",
|
|
"connectionId": "conn-1",
|
|
"selectedMajor": 1,
|
|
"selectedMinor": 0,
|
|
"contractDigest": contract_digest(),
|
|
"limits": Limits::default().to_json(),
|
|
})),
|
|
attachments: Vec::new(),
|
|
}
|
|
.encode()
|
|
.unwrap();
|
|
let mut response = (hello_reply.len() as u32).to_le_bytes().to_vec();
|
|
response.extend(hello_reply);
|
|
let (dropped, transport_dropped) = oneshot::channel();
|
|
let client = Client::connect(
|
|
Transport::from_stream(WriteBlocksAsReaderFails {
|
|
response,
|
|
read: 0,
|
|
hello_written: false,
|
|
command_write_started: false,
|
|
reader_waker: None,
|
|
dropped: Some(dropped),
|
|
}),
|
|
ClientConfig::new("client", dir.path()),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
|
|
let result = tokio::time::timeout(
|
|
Duration::from_secs(1),
|
|
client.declare_topic("t.x", Retained::None),
|
|
)
|
|
.await
|
|
.expect("reader failure left command pending");
|
|
assert_eq!(result.unwrap_err().code, ErrorCode::RouterLost);
|
|
tokio::time::timeout(Duration::from_secs(1), transport_dropped)
|
|
.await
|
|
.expect("reader failure left writer blocked")
|
|
.expect("transport drop signal was discarded");
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn rejected_call_id_still_advances_monotonic_watermark() {
|
|
let e = env(Via::Memory).await;
|
|
let server = e.client("server").await;
|
|
let _service = server
|
|
.register("example.ids", ServiceConfig::default())
|
|
.await
|
|
.unwrap();
|
|
let mut raw = e.raw_hello("caller").await;
|
|
|
|
let rejected = raw
|
|
.call(
|
|
"rpc.call",
|
|
json!({"callId": "call-5", "target": "missing.service", "expectedIncarnation": null, "method": "M", "payload": {}}),
|
|
)
|
|
.await;
|
|
assert_eq!(common::code(&rejected), "NO_SERVICE");
|
|
let decreasing = raw
|
|
.call(
|
|
"rpc.call",
|
|
json!({"callId": "call-4", "target": "example.ids", "expectedIncarnation": null, "method": "M", "payload": {}}),
|
|
)
|
|
.await;
|
|
assert_eq!(common::code(&decreasing), "INVALID_ENVELOPE");
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn unsent_oversized_call_rolls_back_its_local_slot() {
|
|
let limits = Limits {
|
|
max_active_calls_per_client: 1,
|
|
..Limits::default()
|
|
};
|
|
let e = env_with(Via::Memory, limits, Policy::open()).await;
|
|
let server = e.client("server").await;
|
|
let caller = e.client("caller").await;
|
|
let mut service = server
|
|
.register("example.rollback", ServiceConfig::default())
|
|
.await
|
|
.unwrap();
|
|
let oversized = caller
|
|
.call(
|
|
"example.rollback",
|
|
None,
|
|
"Work",
|
|
obj(json!({"blob": "x".repeat(70_000)})),
|
|
&[],
|
|
)
|
|
.await;
|
|
assert_eq!(oversized.unwrap_err().code, ErrorCode::InvalidEnvelope);
|
|
|
|
let mut pending = caller
|
|
.call("example.rollback", None, "Work", obj(json!({})), &[])
|
|
.await
|
|
.unwrap();
|
|
let request = within("request after rollback", service.next())
|
|
.await
|
|
.unwrap();
|
|
request.reply(obj(json!({"ok": true})), &[]).await.unwrap();
|
|
assert_eq!(pending.result().await.unwrap().outcome()["ok"], true);
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn caller_disconnect_cleanup_works_before_and_after_consumption() {
|
|
let e = env(Via::Memory).await;
|
|
let server = e.client("server").await;
|
|
let mut service = server
|
|
.register("example.detach", ServiceConfig::default())
|
|
.await
|
|
.unwrap();
|
|
|
|
let caller = e.client("caller-a").await;
|
|
let pending = caller
|
|
.call("example.detach", None, "Work", obj(json!({})), &[])
|
|
.await
|
|
.unwrap();
|
|
let request = within("request a", service.next()).await.unwrap();
|
|
let responder = request.responder();
|
|
drop(request);
|
|
e.settle("request consumed with responder", |s| {
|
|
s.owners == 0 && s.reply_capabilities == 1
|
|
})
|
|
.await;
|
|
caller.close().await;
|
|
drop(pending);
|
|
// `Client::close` waits for this client to stop; the router marks the call detached when
|
|
// *it* observes the disconnect, in its own connection task. Asserting the reply routing
|
|
// before that is a race: under load the reply reaches the router first and is routed to a
|
|
// connection that is already closing. Nothing escapes -- teardown releases those roots --
|
|
// but `routed` is then true. The contract sentence is about a reply to an already detached
|
|
// call, so the test waits for the teardown it is talking about.
|
|
e.settle("caller-a disconnected", |s| s.connections == 1).await;
|
|
assert!(!responder.reply(obj(json!({})), &[]).await.unwrap());
|
|
drop(responder);
|
|
e.settle("retained responder retired", |s| {
|
|
s.calls == 0 && s.reply_capabilities == 0
|
|
})
|
|
.await;
|
|
|
|
let caller = e.client("caller-b").await;
|
|
let pending = caller
|
|
.call("example.detach", None, "Work", obj(json!({})), &[])
|
|
.await
|
|
.unwrap();
|
|
let request = within("request b", service.next()).await.unwrap();
|
|
drop(request);
|
|
e.settle("capability released first", |s| s.reply_capabilities == 0)
|
|
.await;
|
|
caller.close().await;
|
|
drop(pending);
|
|
e.settle("consumed call retired on disconnect", |s| s.calls == 0)
|
|
.await;
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn dropping_service_with_buffered_requests_retires_every_call() {
|
|
let e = env(Via::Memory).await;
|
|
let server = e.client("server").await;
|
|
let caller = e.client("caller").await;
|
|
let service = server
|
|
.register(
|
|
"example.buffered",
|
|
ServiceConfig {
|
|
max_queued: 4,
|
|
max_in_flight: 4,
|
|
},
|
|
)
|
|
.await
|
|
.unwrap();
|
|
let mut calls = Vec::new();
|
|
for _ in 0..4 {
|
|
calls.push(
|
|
caller
|
|
.call("example.buffered", None, "Work", obj(json!({})), &[])
|
|
.await
|
|
.unwrap(),
|
|
);
|
|
}
|
|
e.settle("requests buffered in SDK", |s| s.reply_capabilities == 4)
|
|
.await;
|
|
drop(service);
|
|
for call in &mut calls {
|
|
let error = within("buffered call failure", call.result())
|
|
.await
|
|
.unwrap_err();
|
|
assert_eq!(
|
|
(error.code, error.dispatch),
|
|
(ErrorCode::NoService, flybus::Dispatch::Dispatched)
|
|
);
|
|
}
|
|
e.settle("buffered calls retired", |s| {
|
|
s.calls == 0 && s.reply_capabilities == 0 && s.owners == 0
|
|
})
|
|
.await;
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn reply_racing_cancel_has_only_the_two_contract_outcomes() {
|
|
let e = env(Via::Memory).await;
|
|
let server = e.client("server").await;
|
|
let caller = e.client("caller").await;
|
|
let mut service = server
|
|
.register("example.reply-race", ServiceConfig::default())
|
|
.await
|
|
.unwrap();
|
|
for _ in 0..32 {
|
|
let mut pending = caller
|
|
.call("example.reply-race", None, "Work", obj(json!({})), &[])
|
|
.await
|
|
.unwrap();
|
|
let request = within("racing request", service.next()).await.unwrap();
|
|
let responder = request.responder();
|
|
let (cancel, reply) = tokio::join!(
|
|
pending.cancel(),
|
|
responder.reply(obj(json!({"ok": true})), &[])
|
|
);
|
|
match (cancel.unwrap(), reply.unwrap()) {
|
|
(CancelState::Completed, true) => {
|
|
assert_eq!(pending.result().await.unwrap().outcome()["ok"], true);
|
|
}
|
|
(CancelState::ExecutionUnknown, false) => {
|
|
assert_eq!(
|
|
pending.result().await.unwrap_err().code,
|
|
ErrorCode::CallGone
|
|
);
|
|
}
|
|
other => panic!("invalid cancel/reply race outcome: {other:?}"),
|
|
}
|
|
drop((request, responder));
|
|
}
|
|
e.settle("race calls retired", |s| {
|
|
s.calls == 0 && s.reply_capabilities == 0 && s.owners == 0
|
|
})
|
|
.await;
|
|
}
|