660 lines
21 KiB
Rust
660 lines
21 KiB
Rust
//! RPC: exclusive registration, pinned incarnations, request/reply, FIFO, bounds,
|
|
//! cancellation, disconnects and an endpoint-side result cache. Every test runs over both the
|
|
//! in-memory transport and a Unix socket.
|
|
|
|
mod common;
|
|
|
|
use std::collections::HashMap;
|
|
use std::io::Write;
|
|
|
|
use common::{Via, code, env, env_with, obj, quiet, sealed, within};
|
|
use flybus::{CancelState, Dispatch, ErrorCode, Grants, Limits, Pattern, Policy, ServiceConfig};
|
|
use serde_json::json;
|
|
|
|
async fn request_reply_roundtrip(via: Via) {
|
|
let e = env(via).await;
|
|
let server = e.client("server").await;
|
|
let caller = e.client("caller").await;
|
|
let mut svc = server
|
|
.register("example.counter", ServiceConfig::default())
|
|
.await
|
|
.unwrap();
|
|
let mut pending = caller
|
|
.call(
|
|
"example.counter",
|
|
Some(svc.incarnation()),
|
|
"Counter.Increment",
|
|
obj(json!({"amount": 1})),
|
|
&[],
|
|
)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(pending.service_incarnation(), svc.incarnation());
|
|
let req = within("request", svc.next()).await.unwrap();
|
|
assert_eq!(req.method(), "Counter.Increment");
|
|
assert_eq!(req.payload()["amount"], 1);
|
|
assert_eq!(req.target(), "example.counter");
|
|
// The router, not the body, says who called.
|
|
assert_eq!(req.caller(), &caller.info().identity);
|
|
assert!(req.reply(obj(json!({"value": 1})), &[]).await.unwrap());
|
|
drop(req);
|
|
let res = within("result", pending.result()).await.unwrap();
|
|
assert_eq!(res.outcome()["value"], 1);
|
|
assert_eq!(res.responder(), &server.info().identity);
|
|
assert_eq!(res.call_id(), pending.call_id());
|
|
drop(res);
|
|
e.settle("everything consumed", |s| s.calls == 0 && s.owners == 0)
|
|
.await;
|
|
}
|
|
|
|
async fn registration_is_exclusive_and_pinned(via: Via) {
|
|
let e = env(via).await;
|
|
let a = e.client("a").await;
|
|
let b = e.client("b").await;
|
|
let caller = e.client("caller").await;
|
|
let first = a
|
|
.register("agent.fly-a", ServiceConfig::default())
|
|
.await
|
|
.unwrap();
|
|
let dup = b
|
|
.register("agent.fly-a", ServiceConfig::default())
|
|
.await
|
|
.unwrap_err();
|
|
assert_eq!(dup.code, ErrorCode::Conflict);
|
|
let old = first.incarnation().to_owned();
|
|
drop(first);
|
|
// Unregistration is asynchronous; wait for the name to come free.
|
|
let second = loop {
|
|
match b.register("agent.fly-a", ServiceConfig::default()).await {
|
|
Ok(s) => break s,
|
|
Err(err) => assert_eq!(err.code, ErrorCode::Conflict),
|
|
}
|
|
};
|
|
assert_ne!(second.incarnation(), old);
|
|
let changed = caller
|
|
.call(
|
|
"agent.fly-a",
|
|
Some(&old),
|
|
"Agent.Prepare",
|
|
obj(json!({})),
|
|
&[],
|
|
)
|
|
.await
|
|
.unwrap_err();
|
|
assert_eq!(changed.code, ErrorCode::TargetChanged);
|
|
assert_eq!(changed.dispatch, Dispatch::NotDispatched);
|
|
let missing = caller
|
|
.call("agent.nobody", None, "Agent.Prepare", obj(json!({})), &[])
|
|
.await
|
|
.unwrap_err();
|
|
assert_eq!(missing.code, ErrorCode::NoService);
|
|
let over = b
|
|
.register(
|
|
"agent.big",
|
|
ServiceConfig {
|
|
max_queued: 17,
|
|
max_in_flight: 1,
|
|
},
|
|
)
|
|
.await
|
|
.unwrap_err();
|
|
assert_eq!(over.code, ErrorCode::QuotaExceeded);
|
|
}
|
|
|
|
async fn authority_is_enforced(via: Via) {
|
|
let policy = Policy::closed()
|
|
.client(
|
|
"server",
|
|
Grants {
|
|
register: vec![Pattern::exact("a.svc")],
|
|
..Grants::default()
|
|
},
|
|
)
|
|
.client(
|
|
"caller",
|
|
Grants {
|
|
call: vec![Pattern::prefix("a.")],
|
|
..Grants::default()
|
|
},
|
|
);
|
|
let e = env_with(via, Limits::default(), policy).await;
|
|
assert_eq!(
|
|
e.try_client("stranger").await.unwrap_err().code,
|
|
ErrorCode::NotAuthorized
|
|
);
|
|
let server = e.client("server").await;
|
|
let caller = e.client("caller").await;
|
|
// One live connection per configured identity.
|
|
assert_eq!(
|
|
e.try_client("caller").await.unwrap_err().code,
|
|
ErrorCode::NotAuthorized
|
|
);
|
|
assert_eq!(
|
|
server
|
|
.register("b.svc", ServiceConfig::default())
|
|
.await
|
|
.unwrap_err()
|
|
.code,
|
|
ErrorCode::NotAuthorized
|
|
);
|
|
let _svc = server
|
|
.register("a.svc", ServiceConfig::default())
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(
|
|
caller
|
|
.register("a.other", ServiceConfig::default())
|
|
.await
|
|
.unwrap_err()
|
|
.code,
|
|
ErrorCode::NotAuthorized
|
|
);
|
|
assert_eq!(
|
|
caller
|
|
.call("b.svc", None, "M", obj(json!({})), &[])
|
|
.await
|
|
.unwrap_err()
|
|
.code,
|
|
ErrorCode::NotAuthorized
|
|
);
|
|
assert_eq!(
|
|
caller
|
|
.declare_topic("a.t", flybus::Retained::None)
|
|
.await
|
|
.unwrap_err()
|
|
.code,
|
|
ErrorCode::NotAuthorized
|
|
);
|
|
assert!(
|
|
caller
|
|
.call("a.svc", None, "M", obj(json!({})), &[])
|
|
.await
|
|
.is_ok()
|
|
);
|
|
// A reconnect must present a new incarnation.
|
|
let mut cfg = e.config("server");
|
|
cfg.client_incarnation = Some(server.info().identity.client_incarnation.clone());
|
|
drop(_svc);
|
|
server.close().await;
|
|
e.settle("server gone", |s| s.connections == 1).await;
|
|
let reused = flybus::Client::connect(e.transport_as("server").await, cfg)
|
|
.await
|
|
.unwrap_err();
|
|
assert_eq!(reused.code, ErrorCode::NotAuthorized);
|
|
assert!(e.try_client("server").await.is_ok());
|
|
}
|
|
|
|
async fn fifo_dispatch_and_out_of_order_completion(via: Via) {
|
|
let e = env(via).await;
|
|
let server = e.client("server").await;
|
|
let caller = e.client("caller").await;
|
|
let mut svc = server
|
|
.register("example.echo", ServiceConfig::default())
|
|
.await
|
|
.unwrap();
|
|
let mut calls = Vec::new();
|
|
for i in 0..6 {
|
|
calls.push(
|
|
caller
|
|
.call("example.echo", None, "Echo", obj(json!({"i": i})), &[])
|
|
.await
|
|
.unwrap(),
|
|
);
|
|
}
|
|
let mut reqs = Vec::new();
|
|
for i in 0..6 {
|
|
let r = within("request", svc.next()).await.unwrap();
|
|
assert_eq!(r.payload()["i"], i, "first dispatch is FIFO per caller");
|
|
reqs.push(r);
|
|
}
|
|
for r in reqs.iter().rev() {
|
|
assert!(
|
|
r.reply(obj(json!({"echo": r.payload()["i"]})), &[])
|
|
.await
|
|
.unwrap()
|
|
);
|
|
}
|
|
drop(reqs);
|
|
for (i, mut c) in calls.into_iter().enumerate() {
|
|
let res = within("result", c.result()).await.unwrap();
|
|
assert_eq!(res.outcome()["echo"], i, "results correlate by call id");
|
|
}
|
|
}
|
|
|
|
async fn service_queue_backpressure(via: Via) {
|
|
let e = env(via).await;
|
|
let server = e.client("server").await;
|
|
let caller = e.client("caller").await;
|
|
let mut svc = server
|
|
.register(
|
|
"example.slow",
|
|
ServiceConfig {
|
|
max_queued: 1,
|
|
max_in_flight: 1,
|
|
},
|
|
)
|
|
.await
|
|
.unwrap();
|
|
let _c1 = caller
|
|
.call("example.slow", None, "Work", obj(json!({"n": 1})), &[])
|
|
.await
|
|
.unwrap();
|
|
let held = within("first request", svc.next()).await.unwrap();
|
|
let _c2 = caller
|
|
.call("example.slow", None, "Work", obj(json!({"n": 2})), &[])
|
|
.await
|
|
.unwrap();
|
|
let full = caller
|
|
.call("example.slow", None, "Work", obj(json!({"n": 3})), &[])
|
|
.await
|
|
.unwrap_err();
|
|
assert_eq!(full.code, ErrorCode::Backpressure);
|
|
assert_eq!(full.dispatch, Dispatch::NotDispatched);
|
|
// In-flight credit returns only when the request delivery is consumed.
|
|
quiet("second request while the first is held", svc.next()).await;
|
|
drop(held);
|
|
let second = within("second request", svc.next()).await.unwrap();
|
|
assert_eq!(second.payload()["n"], 2);
|
|
}
|
|
|
|
async fn cancellation_states(via: Via) {
|
|
let e = env(via).await;
|
|
let server = e.client("server").await;
|
|
let caller = e.client("caller").await;
|
|
let mut svc = server
|
|
.register(
|
|
"example.worker",
|
|
ServiceConfig {
|
|
max_queued: 4,
|
|
max_in_flight: 1,
|
|
},
|
|
)
|
|
.await
|
|
.unwrap();
|
|
|
|
let mut dispatched = caller
|
|
.call("example.worker", None, "Work", obj(json!({"n": 1})), &[])
|
|
.await
|
|
.unwrap();
|
|
let req1 = within("request 1", svc.next()).await.unwrap();
|
|
let mut queued = caller
|
|
.call("example.worker", None, "Work", obj(json!({"n": 2})), &[])
|
|
.await
|
|
.unwrap();
|
|
|
|
assert_eq!(
|
|
queued.cancel().await.unwrap(),
|
|
CancelState::CancelledBeforeDispatch
|
|
);
|
|
let gone = within("cancelled result", queued.result())
|
|
.await
|
|
.unwrap_err();
|
|
assert_eq!(
|
|
(gone.code, gone.dispatch),
|
|
(ErrorCode::CallGone, Dispatch::NotDispatched)
|
|
);
|
|
|
|
assert_eq!(
|
|
dispatched.cancel().await.unwrap(),
|
|
CancelState::ExecutionUnknown
|
|
);
|
|
let unknown = within("detached result", dispatched.result())
|
|
.await
|
|
.unwrap_err();
|
|
assert_eq!(
|
|
(unknown.code, unknown.dispatch),
|
|
(ErrorCode::CallGone, Dispatch::Unknown)
|
|
);
|
|
// The handler still finishes; its reply reaches nobody and is not an error.
|
|
assert!(!req1.reply(obj(json!({"late": true})), &[]).await.unwrap());
|
|
drop(req1);
|
|
|
|
let mut done = caller
|
|
.call("example.worker", None, "Work", obj(json!({"n": 3})), &[])
|
|
.await
|
|
.unwrap();
|
|
let req3 = within("request 3", svc.next()).await.unwrap();
|
|
assert_eq!(
|
|
req3.payload()["n"],
|
|
3,
|
|
"the cancelled call was never dispatched"
|
|
);
|
|
assert!(req3.reply(obj(json!({"ok": 3})), &[]).await.unwrap());
|
|
// Wait for the result to be admitted before cancelling.
|
|
e.settle("result admitted", |s| s.calls == 1).await;
|
|
let state = done.cancel().await.unwrap();
|
|
assert_eq!(state, CancelState::Completed);
|
|
let res = within("completed result", done.result()).await.unwrap();
|
|
assert_eq!(res.outcome()["ok"], 3);
|
|
drop((res, req3));
|
|
e.settle("calls retired", |s| s.calls == 0).await;
|
|
assert_eq!(done.cancel().await.unwrap(), CancelState::CallGone);
|
|
}
|
|
|
|
async fn replies_are_single_and_independent_of_the_request_guard(via: Via) {
|
|
let e = env(via).await;
|
|
let server = e.client("server").await;
|
|
let caller = e.client("caller").await;
|
|
let mut svc = server
|
|
.register("example.once", ServiceConfig::default())
|
|
.await
|
|
.unwrap();
|
|
let mut pending = caller
|
|
.call("example.once", None, "Do", obj(json!({})), &[])
|
|
.await
|
|
.unwrap();
|
|
let req = within("request", svc.next()).await.unwrap();
|
|
let responder = req.responder();
|
|
drop(req); // consumed before replying
|
|
assert!(
|
|
responder
|
|
.reply(obj(json!({"first": true})), &[])
|
|
.await
|
|
.unwrap()
|
|
);
|
|
let again = responder
|
|
.reply(obj(json!({"second": true})), &[])
|
|
.await
|
|
.unwrap_err();
|
|
assert_eq!(again.code, ErrorCode::CallGone);
|
|
let res = within("result", pending.result()).await.unwrap();
|
|
assert_eq!(res.outcome()["first"], true);
|
|
}
|
|
|
|
async fn service_disconnect_fails_calls(via: Via) {
|
|
let e = env(via).await;
|
|
let server = e.client("server").await;
|
|
let caller = e.client("caller").await;
|
|
let mut svc = server
|
|
.register(
|
|
"example.fragile",
|
|
ServiceConfig {
|
|
max_queued: 4,
|
|
max_in_flight: 1,
|
|
},
|
|
)
|
|
.await
|
|
.unwrap();
|
|
let mut c1 = caller
|
|
.call("example.fragile", None, "Do", obj(json!({})), &[])
|
|
.await
|
|
.unwrap();
|
|
let held = within("request", svc.next()).await.unwrap();
|
|
let mut c2 = caller
|
|
.call("example.fragile", None, "Do", obj(json!({})), &[])
|
|
.await
|
|
.unwrap();
|
|
// Keep the first delivery credit occupied until unregister has synchronously failed the
|
|
// queued call. Otherwise consuming it may truthfully dispatch call 2 before unregister.
|
|
drop(svc);
|
|
let r2 = within("queued call", c2.result()).await.unwrap_err();
|
|
assert_eq!(
|
|
(r2.code, r2.dispatch),
|
|
(ErrorCode::NoService, Dispatch::NotDispatched)
|
|
);
|
|
drop(held);
|
|
server.close().await;
|
|
let r1 = within("dispatched call", c1.result()).await.unwrap_err();
|
|
assert_eq!(
|
|
(r1.code, r1.dispatch),
|
|
(ErrorCode::NoService, Dispatch::Dispatched)
|
|
);
|
|
e.settle("nothing left", |s| s.calls == 0 && s.services == 0)
|
|
.await;
|
|
}
|
|
|
|
async fn unregister_fails_queued_but_dispatched_may_reply(via: Via) {
|
|
let e = env(via).await;
|
|
let server = e.client("server").await;
|
|
let caller = e.client("caller").await;
|
|
let mut svc = server
|
|
.register(
|
|
"example.leaving",
|
|
ServiceConfig {
|
|
max_queued: 4,
|
|
max_in_flight: 1,
|
|
},
|
|
)
|
|
.await
|
|
.unwrap();
|
|
let mut c1 = caller
|
|
.call("example.leaving", None, "Do", obj(json!({})), &[])
|
|
.await
|
|
.unwrap();
|
|
let req = within("request", svc.next()).await.unwrap();
|
|
let mut c2 = caller
|
|
.call("example.leaving", None, "Do", obj(json!({})), &[])
|
|
.await
|
|
.unwrap();
|
|
drop(svc);
|
|
let r2 = within("queued call", c2.result()).await.unwrap_err();
|
|
assert_eq!(
|
|
(r2.code, r2.dispatch),
|
|
(ErrorCode::NoService, Dispatch::NotDispatched)
|
|
);
|
|
assert!(req.reply(obj(json!({"done": true})), &[]).await.unwrap());
|
|
assert_eq!(
|
|
within("dispatched call", c1.result())
|
|
.await
|
|
.unwrap()
|
|
.outcome()["done"],
|
|
true
|
|
);
|
|
}
|
|
|
|
async fn raw_call_ids_and_forged_replies(via: Via) {
|
|
let e = env(via).await;
|
|
let server = e.client("server").await;
|
|
let mut svc = server
|
|
.register("example.raw", ServiceConfig::default())
|
|
.await
|
|
.unwrap();
|
|
let mut raw = e.raw_hello("rawcaller").await;
|
|
let call = |id: &str| json!({"callId": id, "target": "example.raw", "expectedIncarnation": null, "method": "M", "payload": {}});
|
|
assert_eq!(code(&raw.call("rpc.call", call("call-5")).await), "OK");
|
|
assert_eq!(
|
|
code(&raw.call("rpc.call", call("call-5")).await),
|
|
"INVALID_ENVELOPE"
|
|
);
|
|
assert_eq!(
|
|
code(&raw.call("rpc.call", call("call-3")).await),
|
|
"INVALID_ENVELOPE"
|
|
);
|
|
assert_eq!(
|
|
code(&raw.call("rpc.call", call("call-05")).await),
|
|
"INVALID_ENVELOPE"
|
|
);
|
|
// Identity comes from the connection; a body cannot claim one.
|
|
let mut forged = call("call-6");
|
|
forged["caller"] = json!({"clientId": "server", "clientIncarnation": "x"});
|
|
assert_eq!(
|
|
code(&raw.call("rpc.call", forged).await),
|
|
"INVALID_ENVELOPE"
|
|
);
|
|
assert_eq!(code(&raw.call("rpc.call", call("call-6")).await), "OK");
|
|
let req = within("request", svc.next()).await.unwrap();
|
|
assert_eq!(req.caller().client_id, "rawcaller");
|
|
// A third party cannot answer someone else's request.
|
|
let mut other = e.raw_hello("intruder").await;
|
|
let r = other
|
|
.call(
|
|
"rpc.reply",
|
|
json!({"callId": req.call_id(), "requestDeliveryId": req.delivery_id(), "outcome": {}}),
|
|
)
|
|
.await;
|
|
assert_eq!(code(&r), "OWNER_INVALID");
|
|
assert!(req.reply(obj(json!({"real": true})), &[]).await.unwrap());
|
|
let result = raw.event().await;
|
|
assert_eq!(result.op, "rpc.result");
|
|
assert_eq!(result.body["outcome"]["real"], true);
|
|
assert_eq!(result.body["responder"]["clientId"], "server");
|
|
}
|
|
|
|
/// bus-v1 section 6: an endpoint caches Artifact handles plus payload; a domain retry with a
|
|
/// fresh call id gets fresh delivery ownership over the same immutable bytes.
|
|
async fn endpoint_cache_replays_artifact_results(via: Via) {
|
|
let e = env(via).await;
|
|
let server = e.client("server").await;
|
|
let caller = e.client("caller").await;
|
|
let mut svc = server
|
|
.register("agent.cached", ServiceConfig::default())
|
|
.await
|
|
.unwrap();
|
|
let service = tokio::spawn(async move {
|
|
let mut cache: HashMap<String, flybus::Artifact> = HashMap::new();
|
|
let mut executions = 0;
|
|
while let Some(req) = svc.next().await {
|
|
if req.method() == "Cache.Evict" {
|
|
cache.clear();
|
|
req.reply(obj(json!({})), &[]).await.unwrap();
|
|
continue;
|
|
}
|
|
let rid = req.payload()["requestId"].as_str().unwrap().to_owned();
|
|
if !cache.contains_key(&rid) {
|
|
executions += 1;
|
|
let bytes = format!("result of {rid}").into_bytes();
|
|
let mut w = server
|
|
.artifacts()
|
|
.allocate(bytes.len() as u64, "text/plain")
|
|
.await
|
|
.unwrap();
|
|
w.write_all(&bytes).unwrap();
|
|
cache.insert(rid.clone(), w.seal().await.unwrap());
|
|
}
|
|
let art = cache[&rid].clone();
|
|
req.reply(obj(json!({"executions": executions})), &[("state", &art)])
|
|
.await
|
|
.unwrap();
|
|
}
|
|
});
|
|
let mut ids = Vec::new();
|
|
for _ in 0..2 {
|
|
let res = within(
|
|
"result",
|
|
caller.call_and_wait(
|
|
"agent.cached",
|
|
None,
|
|
"Agent.Prepare",
|
|
obj(json!({"requestId": "req-41"})),
|
|
&[],
|
|
),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(
|
|
res.outcome()["executions"],
|
|
1,
|
|
"the retry did not re-execute"
|
|
);
|
|
let art = res.artifact("state").unwrap();
|
|
assert_eq!(art.read_all().await.unwrap(), b"result of req-41");
|
|
ids.push((
|
|
res.delivery_id().to_owned(),
|
|
art.reference().artifact_id.clone(),
|
|
));
|
|
}
|
|
assert_ne!(ids[0].0, ids[1].0, "each replay is a fresh delivery");
|
|
assert_eq!(ids[0].1, ids[1].1, "of the same immutable object");
|
|
e.settle("cache holds the only root", |s| {
|
|
s.sealed_artifacts == 1 && s.artifact_roots == 1
|
|
})
|
|
.await;
|
|
caller
|
|
.call_and_wait("agent.cached", None, "Cache.Evict", obj(json!({})), &[])
|
|
.await
|
|
.unwrap();
|
|
e.settle("eviction collects", |s| {
|
|
s.artifacts == 0 && s.store_bytes == 0
|
|
})
|
|
.await;
|
|
e.settle_files("sealed", 0).await;
|
|
service.abort();
|
|
}
|
|
|
|
async fn dropped_call_is_cancelled_and_late_result_consumed(via: Via) {
|
|
let e = env(via).await;
|
|
let server = e.client("server").await;
|
|
let caller = e.client("caller").await;
|
|
let mut svc = server
|
|
.register("example.abandon", ServiceConfig::default())
|
|
.await
|
|
.unwrap();
|
|
let art = sealed(&server, b"payload", "text/plain").await;
|
|
let pending = caller
|
|
.call("example.abandon", None, "Do", obj(json!({})), &[])
|
|
.await
|
|
.unwrap();
|
|
let req = within("request", svc.next()).await.unwrap();
|
|
drop(pending);
|
|
e.settle("call detached", |s| s.calls == 1 && s.active_calls == 0)
|
|
.await;
|
|
// Detached: the reply is not routed and creates no caller-side roots.
|
|
let routed = req.reply(obj(json!({})), &[("a", &art)]).await.unwrap();
|
|
assert!(!routed);
|
|
drop((req, art));
|
|
e.settle("nothing retained", |s| {
|
|
s.calls == 0 && s.artifacts == 0 && s.owners == 0
|
|
})
|
|
.await;
|
|
|
|
// A result that arrives after its caller stopped waiting is consumed by the reactor.
|
|
let pending = caller
|
|
.call("example.abandon", None, "Do", obj(json!({})), &[])
|
|
.await
|
|
.unwrap();
|
|
let req = within("request", svc.next()).await.unwrap();
|
|
let art = sealed(&server, b"late", "text/plain").await;
|
|
req.reply(obj(json!({})), &[("a", &art)]).await.unwrap();
|
|
drop((req, art));
|
|
e.settle("result delivered", |s| s.calls == 1 && s.owners == 1)
|
|
.await;
|
|
drop(pending);
|
|
e.settle("result consumed", |s| {
|
|
s.calls == 0 && s.artifacts == 0 && s.owners == 0
|
|
})
|
|
.await;
|
|
assert_eq!(caller.control_errors(), 0);
|
|
}
|
|
|
|
async fn active_call_limit(via: Via) {
|
|
let limits = Limits {
|
|
max_active_calls_per_client: 2,
|
|
..Limits::default()
|
|
};
|
|
let e = env_with(via, limits, Policy::open()).await;
|
|
let server = e.client("server").await;
|
|
let caller = e.client("caller").await;
|
|
let _svc = server
|
|
.register("example.limit", ServiceConfig::default())
|
|
.await
|
|
.unwrap();
|
|
let _a = caller
|
|
.call("example.limit", None, "M", obj(json!({})), &[])
|
|
.await
|
|
.unwrap();
|
|
let _b = caller
|
|
.call("example.limit", None, "M", obj(json!({})), &[])
|
|
.await
|
|
.unwrap();
|
|
let c = caller
|
|
.call("example.limit", None, "M", obj(json!({})), &[])
|
|
.await
|
|
.unwrap_err();
|
|
assert_eq!(c.code, ErrorCode::Backpressure);
|
|
}
|
|
|
|
both_transports!(
|
|
request_reply_roundtrip,
|
|
registration_is_exclusive_and_pinned,
|
|
authority_is_enforced,
|
|
fifo_dispatch_and_out_of_order_completion,
|
|
service_queue_backpressure,
|
|
cancellation_states,
|
|
replies_are_single_and_independent_of_the_request_guard,
|
|
service_disconnect_fails_calls,
|
|
unregister_fails_queued_but_dispatched_may_reply,
|
|
raw_call_ids_and_forged_replies,
|
|
endpoint_cache_replays_artifact_results,
|
|
dropped_call_is_cancelled_and_late_result_consumed,
|
|
active_call_limit,
|
|
);
|