355 lines
13 KiB
Rust
355 lines
13 KiB
Rust
//! Wire and connection: hello negotiation, strict frame and envelope validation, body errors
|
|
//! that keep the connection, envelope size limits and control-lane exhaustion.
|
|
|
|
mod common;
|
|
|
|
use std::time::Duration;
|
|
|
|
use common::{Raw, Via, code, env, env_with, obj};
|
|
use flybus::wire::{Kind, MAX_ENVELOPE_BYTES, contract_digest};
|
|
use flybus::{ErrorCode, Limits, Policy, Retained};
|
|
use serde_json::{Value, json};
|
|
|
|
async fn hello_negotiation(via: Via) {
|
|
let e = env(via).await;
|
|
let mut raw = e.raw().await;
|
|
let v = raw.hello("probe").await.unwrap();
|
|
assert_eq!(v["routerId"], e.router.router_id());
|
|
assert_eq!(
|
|
(v["selectedMajor"].as_u64(), v["selectedMinor"].as_u64()),
|
|
(Some(1), Some(0))
|
|
);
|
|
assert_eq!(v["contractDigest"], contract_digest());
|
|
assert!(v["connectionId"].as_str().unwrap().starts_with("conn-"));
|
|
assert_eq!(
|
|
flybus::Limits::from_json(&v["limits"]).unwrap(),
|
|
Limits::default()
|
|
);
|
|
let c = e.client("sdk").await;
|
|
assert_eq!(c.info().router_id, e.router.router_id());
|
|
assert_ne!(c.info().connection_id, v["connectionId"]);
|
|
}
|
|
|
|
async fn expect_closed(raw: &mut Raw, want: &str) {
|
|
let notice = raw
|
|
.closing()
|
|
.await
|
|
.unwrap_or_else(|| panic!("no connection.closing notice (wanted {want})"));
|
|
assert_eq!(notice["code"], want, "{notice:?}");
|
|
}
|
|
|
|
async fn hello_refusals(via: Via) {
|
|
let e = env_with(
|
|
via,
|
|
Limits::default(),
|
|
Policy::closed().client("known", flybus::Grants::all()),
|
|
)
|
|
.await;
|
|
let mut raw = e.raw_as("known").await;
|
|
raw.command(
|
|
"topic.declare",
|
|
json!({"name": "t.x", "retained": "none"}),
|
|
json!([]),
|
|
)
|
|
.await;
|
|
expect_closed(&mut raw, "INVALID_ENVELOPE").await;
|
|
|
|
let mut raw = e.raw_as("known").await;
|
|
let r = raw
|
|
.call(
|
|
"bus.hello",
|
|
json!({"clientId": "known", "clientIncarnation": "i-1", "supportedMajors": [2, 3]}),
|
|
)
|
|
.await;
|
|
assert_eq!(code(&r), "VERSION_MISMATCH");
|
|
assert!(
|
|
raw.recv().await.is_none(),
|
|
"refused hello closes the connection"
|
|
);
|
|
|
|
let mut raw = e.raw_as("known").await;
|
|
assert_eq!(code(&raw.hello("stranger").await), "NOT_AUTHORIZED");
|
|
assert!(raw.recv().await.is_none());
|
|
|
|
let mut raw = e.raw_as("known").await;
|
|
let r = raw.call("bus.hello", json!({"clientId": "known", "clientIncarnation": "i-2", "supportedMajors": [1], "admin": true})).await;
|
|
assert_eq!(code(&r), "INVALID_ENVELOPE");
|
|
|
|
let mut raw = e.raw_as("known").await;
|
|
raw.hello("known").await.unwrap();
|
|
raw.command(
|
|
"bus.hello",
|
|
json!({"clientId": "known", "clientIncarnation": "i-3", "supportedMajors": [1]}),
|
|
json!([]),
|
|
)
|
|
.await;
|
|
expect_closed(&mut raw, "INVALID_ENVELOPE").await;
|
|
e.settle("all refused connections released", |s| s.connections == 0)
|
|
.await;
|
|
}
|
|
|
|
fn envelope(id: &str, body: Value) -> Value {
|
|
json!({"protocol": "flybus", "major": 1, "minor": 0, "id": id, "replyTo": null, "kind": "command",
|
|
"op": "publish", "body": body, "attachments": []})
|
|
}
|
|
|
|
async fn malformed_frames_close_the_connection(via: Via) {
|
|
let e = env(via).await;
|
|
let admin = e.client("admin").await;
|
|
admin.declare_topic("t.x", Retained::None).await.unwrap();
|
|
let publish = |id: &str| envelope(id, json!({"topic": "t.x", "payload": {}}));
|
|
let text = |v: Value| serde_json::to_string(&v).unwrap();
|
|
let att = |name: &str| {
|
|
json!({"name": name, "ref": {"storeId": "s", "artifactId": "a-1", "generation": "1",
|
|
"byteLength": "1", "contentType": "x", "digest": null}, "ownerId": "own-1"})
|
|
};
|
|
let mut too_many = publish("msg-2");
|
|
too_many["attachments"] = Value::Array((0..33).map(|i| att(&format!("a{i}"))).collect());
|
|
let mut duplicate_names = publish("msg-2");
|
|
duplicate_names["attachments"] = json!([att("a"), att("a")]);
|
|
let mut unknown_field = publish("msg-2");
|
|
unknown_field["priority"] = json!(1);
|
|
let mut from_router = publish("msg-2");
|
|
from_router["kind"] = json!("reply");
|
|
let mut reply_to = publish("msg-2");
|
|
reply_to["replyTo"] = json!("msg-1");
|
|
let mut major = publish("msg-2");
|
|
major["major"] = json!(2);
|
|
let mut not_flybus = publish("msg-2");
|
|
not_flybus["protocol"] = json!("flybusx");
|
|
|
|
let cases: Vec<(&str, Vec<u8>)> = vec![
|
|
("nested duplicate key", br#"{"protocol":"flybus","major":1,"minor":0,"id":"msg-2","replyTo":null,"kind":"command","op":"publish","body":{"topic":"t.x","payload":{"a":{"b":1,"b":2}}},"attachments":[]}"#.to_vec()),
|
|
("top-level duplicate key", br#"{"protocol":"flybus","major":1,"minor":0,"id":"msg-2","id":"msg-3","replyTo":null,"kind":"command","op":"publish","body":{"topic":"t.x","payload":{}},"attachments":[]}"#.to_vec()),
|
|
("NaN", br#"{"protocol":"flybus","major":1,"minor":0,"id":"msg-2","replyTo":null,"kind":"command","op":"publish","body":{"topic":"t.x","payload":{"v":NaN}},"attachments":[]}"#.to_vec()),
|
|
("invalid UTF-8", [&text(publish("msg-2")).into_bytes()[..60], b"\xff\xfe", &text(publish("msg-2")).into_bytes()[62..]].concat()),
|
|
("trailing bytes", [text(publish("msg-2")).into_bytes(), b" {}".to_vec()].concat()),
|
|
("not an object", b"[1,2,3]".to_vec()),
|
|
("non-canonical id", text(publish("msg-02")).into_bytes()),
|
|
("too many attachments", text(too_many).into_bytes()),
|
|
("duplicate attachment names", text(duplicate_names).into_bytes()),
|
|
("unknown envelope field", text(unknown_field).into_bytes()),
|
|
("reply kind from a client", text(from_router).into_bytes()),
|
|
("non-null replyTo", text(reply_to).into_bytes()),
|
|
("wrong major after hello", text(major).into_bytes()),
|
|
("wrong protocol", text(not_flybus).into_bytes()),
|
|
];
|
|
for (i, (what, bytes)) in cases.into_iter().enumerate() {
|
|
let mut raw = e.raw_hello(&format!("bad-{i}")).await;
|
|
raw.send_bytes(&bytes).await;
|
|
let notice = raw
|
|
.closing()
|
|
.await
|
|
.unwrap_or_else(|| panic!("{what}: no closing notice"));
|
|
assert_eq!(notice["code"], "INVALID_ENVELOPE", "{what}");
|
|
e.settle(what, |s| s.connections == 1).await;
|
|
}
|
|
|
|
// Ids must increase.
|
|
let mut raw = e.raw_hello("replay").await;
|
|
raw.send_bytes(&serde_json::to_vec(&publish("msg-5")).unwrap())
|
|
.await;
|
|
raw.send_bytes(&serde_json::to_vec(&publish("msg-5")).unwrap())
|
|
.await;
|
|
expect_closed(&mut raw, "INVALID_ENVELOPE").await;
|
|
|
|
// Framing: a zero length, and a length over the limit with no body behind it.
|
|
let mut raw = e.raw_hello("zero").await;
|
|
raw.send_prefix(0).await;
|
|
expect_closed(&mut raw, "INVALID_ENVELOPE").await;
|
|
let mut raw = e.raw_hello("huge").await;
|
|
raw.send_prefix(u32::MAX).await;
|
|
expect_closed(&mut raw, "INVALID_ENVELOPE").await;
|
|
// A frame cut short by the end of the stream just ends the connection.
|
|
let mut raw = e.raw_hello("short").await;
|
|
raw.send_prefix(100).await;
|
|
drop(raw);
|
|
e.settle("only the admin remains", |s| s.connections == 1)
|
|
.await;
|
|
}
|
|
|
|
async fn body_errors_keep_the_connection(via: Via) {
|
|
let e = env(via).await;
|
|
let mut raw = e.raw_hello("raw").await;
|
|
let cases = [
|
|
("no.such.op", json!({}), json!([])),
|
|
(
|
|
"topic.declare",
|
|
json!({"name": "t.a", "retained": "none", "extra": 1}),
|
|
json!([]),
|
|
),
|
|
(
|
|
"topic.declare",
|
|
json!({"name": "t..a", "retained": "none"}),
|
|
json!([]),
|
|
),
|
|
(
|
|
"topic.declare",
|
|
json!({"name": "t.a", "retained": "sometimes"}),
|
|
json!([]),
|
|
),
|
|
("topic.declare", json!({"name": "t.a"}), json!([])),
|
|
(
|
|
"subscribe",
|
|
json!({"topic": "t.a", "mode": "bounded", "maxQueued": 0, "maxInFlight": 1, "replayLatest": false}),
|
|
json!([]),
|
|
),
|
|
(
|
|
"subscribe",
|
|
json!({"topic": "t.a", "mode": "bounded", "maxQueued": 1.0, "maxInFlight": 1, "replayLatest": false}),
|
|
json!([]),
|
|
),
|
|
(
|
|
"subscribe",
|
|
json!({"topic": "t.a", "mode": "bounded", "maxQueued": 65536, "maxInFlight": 1, "replayLatest": false}),
|
|
json!([]),
|
|
),
|
|
(
|
|
"artifact.allocate",
|
|
json!({"byteLength": 5, "contentType": "x"}),
|
|
json!([]),
|
|
),
|
|
(
|
|
"artifact.allocate",
|
|
json!({"byteLength": "18446744073709551616", "contentType": "x"}),
|
|
json!([]),
|
|
),
|
|
(
|
|
"rpc.call",
|
|
json!({"callId": "call-1", "target": "a.b", "expectedIncarnation": null, "method": "", "payload": {}}),
|
|
json!([]),
|
|
),
|
|
(
|
|
"rpc.call",
|
|
json!({"callId": "call-1", "target": "a.b", "expectedIncarnation": null, "method": "M", "payload": []}),
|
|
json!([]),
|
|
),
|
|
(
|
|
"delivery.consumed",
|
|
json!({"deliveryIds": "dlv-1"}),
|
|
json!([]),
|
|
),
|
|
(
|
|
"topic.declare",
|
|
json!({"name": "t.a", "retained": "none"}),
|
|
json!([{"name": "x", "ref": {"storeId": "s", "artifactId": "a-1", "generation": "1", "byteLength": "1", "contentType": "x", "digest": null}, "ownerId": "own-1"}]),
|
|
),
|
|
];
|
|
for (op, body, atts) in cases {
|
|
let r = raw.call_with(op, body.clone(), atts).await;
|
|
assert_eq!(
|
|
r,
|
|
Err(("INVALID_ENVELOPE".into(), "not-dispatched".into())),
|
|
"{op} {body}"
|
|
);
|
|
}
|
|
assert_eq!(
|
|
code(
|
|
&raw.call("topic.declare", json!({"name": "t.a", "retained": "none"}))
|
|
.await
|
|
),
|
|
"OK"
|
|
);
|
|
assert_eq!(e.stats().connections, 1);
|
|
}
|
|
|
|
async fn envelope_size_limits(via: Via) {
|
|
let e = env(via).await;
|
|
// Exactly 65536 bytes is a frame the router reads (and answers); one more is not.
|
|
let sized = |id: &str, total: usize| {
|
|
let base = serde_json::to_vec(
|
|
&json!({"protocol": "flybus", "major": 1, "minor": 0, "id": id, "replyTo": null,
|
|
"kind": "command", "op": "no.such.op", "body": {"pad": ""}, "attachments": []}),
|
|
)
|
|
.unwrap();
|
|
let pad = "p".repeat(total - base.len());
|
|
let bytes = serde_json::to_vec(
|
|
&json!({"protocol": "flybus", "major": 1, "minor": 0, "id": id, "replyTo": null,
|
|
"kind": "command", "op": "no.such.op", "body": {"pad": pad}, "attachments": []}),
|
|
)
|
|
.unwrap();
|
|
assert_eq!(bytes.len(), total);
|
|
bytes
|
|
};
|
|
let mut raw = e.raw_hello("raw").await;
|
|
raw.send_bytes(&sized("msg-2", MAX_ENVELOPE_BYTES)).await;
|
|
assert_eq!(code(&raw.reply("msg-2").await), "INVALID_ENVELOPE");
|
|
raw.send_bytes(&sized("msg-3", MAX_ENVELOPE_BYTES + 1))
|
|
.await;
|
|
expect_closed(&mut raw, "INVALID_ENVELOPE").await;
|
|
|
|
// The SDK refuses to send an oversized envelope and the connection lives on.
|
|
let c = e.client("c").await;
|
|
c.declare_topic("t.big", Retained::None).await.unwrap();
|
|
let huge = c
|
|
.publish(
|
|
"t.big",
|
|
obj(json!({"blob": "x".repeat(MAX_ENVELOPE_BYTES)})),
|
|
&[],
|
|
)
|
|
.await
|
|
.unwrap_err();
|
|
assert_eq!(huge.code, ErrorCode::InvalidEnvelope);
|
|
// An envelope that fits but whose delivery (with its added ids) would not is refused at
|
|
// admission rather than truncated later.
|
|
let edge = c
|
|
.publish("t.big", obj(json!({"blob": "x".repeat(65_300)})), &[])
|
|
.await
|
|
.unwrap_err();
|
|
assert_eq!(edge.code, ErrorCode::InvalidEnvelope);
|
|
assert!(edge.message.contains("delivery"), "{edge}");
|
|
assert_eq!(
|
|
c.publish("t.big", obj(json!({"blob": "x".repeat(60_000)})), &[])
|
|
.await
|
|
.unwrap()
|
|
.topic_sequence,
|
|
1
|
|
);
|
|
}
|
|
|
|
/// A client that sends commands and never reads replies is closed once its control lane is
|
|
/// full, with a notice, instead of the router buffering without bound.
|
|
async fn control_lane_exhaustion_closes(via: Via) {
|
|
let limits = Limits {
|
|
max_control_frames: 4,
|
|
..Limits::default()
|
|
};
|
|
let e = env_with(via, limits, Policy::open()).await;
|
|
let mut raw = e.raw_hello("flood").await;
|
|
let mut writer = raw.take_writer();
|
|
let flood = tokio::spawn(async move {
|
|
for i in 2..20_000u64 {
|
|
let env = json!({"protocol": "flybus", "major": 1, "minor": 0, "id": format!("msg-{i}"), "replyTo": null,
|
|
"kind": "command", "op": "no.such.op", "body": {}, "attachments": []});
|
|
if !writer.send(&serde_json::to_vec(&env).unwrap()).await {
|
|
break;
|
|
}
|
|
}
|
|
});
|
|
tokio::time::sleep(Duration::from_millis(300)).await;
|
|
let mut replies = 0;
|
|
let mut last = None;
|
|
while let Some(env) = raw.recv().await {
|
|
match env.kind {
|
|
Kind::Reply => replies += 1,
|
|
_ => last = Some(env),
|
|
}
|
|
}
|
|
let last = last.expect("a closing notice");
|
|
assert_eq!(
|
|
(last.op.as_str(), last.body["code"].as_str()),
|
|
("connection.closing", Some("QUOTA_EXCEEDED"))
|
|
);
|
|
assert!(replies < 19_998, "the router stopped answering");
|
|
flood.abort();
|
|
e.settle("flooder released", |s| s.connections == 0).await;
|
|
}
|
|
|
|
both_transports!(
|
|
hello_negotiation,
|
|
hello_refusals,
|
|
malformed_frames_close_the_connection,
|
|
body_errors_keep_the_connection,
|
|
envelope_size_limits,
|
|
control_lane_exhaustion_closes,
|
|
);
|