704 lines
23 KiB
Rust
704 lines
23 KiB
Rust
//! Artifacts: allocate/seal/read, immutability against live writable handles, ownership
|
|
//! through fan-out and retention, quotas, atomic admission, watermarked releases, abandoned
|
|
//! futures, disconnects and router restarts.
|
|
|
|
mod common;
|
|
|
|
use std::io::{Seek, SeekFrom, Write};
|
|
use std::os::unix::fs::PermissionsExt;
|
|
use std::time::Duration;
|
|
|
|
use common::{Via, code, env, env_with, obj, quiet, sealed, within};
|
|
use flybus::{
|
|
ErrorCode, Limits, Policy, Retained, Router, RouterConfig, ServiceConfig, SubscriptionConfig,
|
|
};
|
|
use serde_json::{Value, json};
|
|
use sha2::{Digest, Sha256};
|
|
|
|
fn sha(bytes: &[u8]) -> String {
|
|
Sha256::digest(bytes)
|
|
.iter()
|
|
.map(|b| format!("{b:02x}"))
|
|
.collect()
|
|
}
|
|
|
|
async fn allocate_write_seal_read(via: Via) {
|
|
let e = env(via).await;
|
|
let c = e.client("c").await;
|
|
let data: Vec<u8> = (0..100_000u32).map(|i| (i % 251) as u8).collect();
|
|
let mut w = c
|
|
.artifacts()
|
|
.allocate(data.len() as u64, "application/octet-stream")
|
|
.await
|
|
.unwrap();
|
|
assert!(
|
|
w.write_all(&[0; 100_001]).is_err(),
|
|
"writes past the allocation are refused"
|
|
);
|
|
w.write_all(&data).unwrap();
|
|
assert_eq!(e.files("staging"), 1);
|
|
let art = w.seal_with_digest(Some(sha(&data))).await.unwrap();
|
|
let r = art.reference();
|
|
assert_eq!(
|
|
(r.generation, r.byte_length, r.digest.clone()),
|
|
(1, 100_000, Some(sha(&data)))
|
|
);
|
|
assert_eq!(r.store_id, e.router.store_id());
|
|
assert_eq!(art.read_all().await.unwrap(), data);
|
|
assert_eq!((e.files("staging"), e.files("sealed")), (0, 1));
|
|
let meta = std::fs::metadata(e.router.store_dir().join("sealed").join(&r.artifact_id)).unwrap();
|
|
assert_eq!(
|
|
meta.permissions().mode() & 0o222,
|
|
0,
|
|
"sealed files are read-only"
|
|
);
|
|
|
|
let empty = c
|
|
.artifacts()
|
|
.allocate(0, "text/plain")
|
|
.await
|
|
.unwrap()
|
|
.seal()
|
|
.await
|
|
.unwrap();
|
|
assert!(empty.read_all().await.unwrap().is_empty());
|
|
assert_eq!(
|
|
c.artifacts().allocate(1, "").await.unwrap_err().code,
|
|
ErrorCode::InvalidEnvelope
|
|
);
|
|
drop((art, empty));
|
|
e.settle("collected", |s| s.artifacts == 0 && s.store_bytes == 0)
|
|
.await;
|
|
e.settle_files("sealed", 0).await;
|
|
}
|
|
|
|
async fn unsealed_artifacts_cannot_be_used(via: Via) {
|
|
let e = env(via).await;
|
|
let mut raw = e.raw_hello("raw").await;
|
|
assert_eq!(
|
|
code(
|
|
&raw.call("topic.declare", json!({"name": "t.x", "retained": "none"}))
|
|
.await
|
|
),
|
|
"OK"
|
|
);
|
|
let (a, _) = raw.allocate(8).await;
|
|
let store = a["writeLocation"]["storeId"].clone();
|
|
let reference = json!({"storeId": store, "artifactId": a["artifactId"], "generation": "1", "byteLength": "8",
|
|
"contentType": "application/octet-stream", "digest": null});
|
|
let att = json!([{"name": "x", "ref": reference, "ownerId": a["ownerId"]}]);
|
|
let r = raw
|
|
.call_with("publish", json!({"topic": "t.x", "payload": {}}), att)
|
|
.await;
|
|
assert_eq!(code(&r), "ARTIFACT_UNSEALED");
|
|
let r = raw
|
|
.call(
|
|
"artifact.open",
|
|
json!({"ref": reference, "ownerId": a["ownerId"]}),
|
|
)
|
|
.await;
|
|
assert_eq!(code(&r), "ARTIFACT_UNSEALED");
|
|
let r = raw
|
|
.call(
|
|
"artifact.retain",
|
|
json!({"ref": reference, "ownerId": a["ownerId"]}),
|
|
)
|
|
.await;
|
|
assert_eq!(code(&r), "ARTIFACT_UNSEALED");
|
|
assert_eq!(code(&raw.seal(&a, Value::Null).await), "OK");
|
|
assert_eq!(
|
|
code(&raw.seal(&a, Value::Null).await),
|
|
"OWNER_INVALID",
|
|
"a writer seals once"
|
|
);
|
|
assert_eq!(
|
|
code(
|
|
&raw.call(
|
|
"artifact.allocate",
|
|
json!({"byteLength": "01", "contentType": "x"})
|
|
)
|
|
.await
|
|
),
|
|
"INVALID_ENVELOPE"
|
|
);
|
|
}
|
|
|
|
/// A producer that kept (or duplicated) its writable descriptor cannot change sealed bytes.
|
|
async fn seal_is_immune_to_live_writable_handles(via: Via) {
|
|
let e = env(via).await;
|
|
let mut raw = e.raw_hello("raw").await;
|
|
let (a, path) = raw.allocate(8).await;
|
|
let mut kept = std::fs::OpenOptions::new().write(true).open(&path).unwrap();
|
|
kept.write_all(b"AAAAAAAA").unwrap();
|
|
let mut dup = kept.try_clone().unwrap();
|
|
let sealed = raw.seal(&a, json!(sha(b"AAAAAAAA"))).await.unwrap();
|
|
// Both descriptors still write, into an inode the store no longer uses.
|
|
dup.seek(SeekFrom::Start(0)).unwrap();
|
|
let _ = dup.write_all(b"BBBBBBBB");
|
|
let _ = kept.write_all(b"CCCC");
|
|
assert!(!path.exists(), "staging is unlinked after sealing");
|
|
let open = raw
|
|
.call(
|
|
"artifact.open",
|
|
json!({"ref": sealed["ref"], "ownerId": sealed["ownerId"]}),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
let read_path = raw.path(&open["readLocation"]);
|
|
assert_eq!(std::fs::read(&read_path).unwrap(), b"AAAAAAAA");
|
|
assert_eq!(
|
|
std::fs::metadata(&read_path).unwrap().permissions().mode() & 0o222,
|
|
0
|
|
);
|
|
}
|
|
|
|
async fn seal_checks_length_and_digest(via: Via) {
|
|
let e = env(via).await;
|
|
let mut raw = e.raw_hello("raw").await;
|
|
let (short, path) = raw.allocate(4).await;
|
|
std::fs::OpenOptions::new()
|
|
.write(true)
|
|
.open(&path)
|
|
.unwrap()
|
|
.set_len(2)
|
|
.unwrap();
|
|
assert_eq!(
|
|
code(&raw.seal(&short, Value::Null).await),
|
|
"ARTIFACT_MISMATCH"
|
|
);
|
|
let (long, path) = raw.allocate(4).await;
|
|
std::fs::OpenOptions::new()
|
|
.write(true)
|
|
.open(&path)
|
|
.unwrap()
|
|
.set_len(6)
|
|
.unwrap();
|
|
assert_eq!(
|
|
code(&raw.seal(&long, Value::Null).await),
|
|
"ARTIFACT_MISMATCH"
|
|
);
|
|
let (bad, _) = raw.allocate(4).await;
|
|
assert_eq!(
|
|
code(&raw.seal(&bad, json!("XYZ")).await),
|
|
"INVALID_ENVELOPE"
|
|
);
|
|
assert_eq!(
|
|
code(&raw.seal(&bad, json!("0".repeat(64))).await),
|
|
"ARTIFACT_MISMATCH"
|
|
);
|
|
// Failed seals clean up both the staging file and the copy.
|
|
let s = e.settle("failed seals cleaned", |s| s.artifacts == 0).await;
|
|
assert_eq!((s.store_bytes, s.owners), (0, 0));
|
|
e.settle_files("staging", 0).await;
|
|
e.settle_files("sealed", 0).await;
|
|
|
|
let c = e.client("c").await;
|
|
let mut w = c.artifacts().allocate(3, "text/plain").await.unwrap();
|
|
w.write_all(b"abc").unwrap();
|
|
let err = w.seal_with_digest(Some(sha(b"abd"))).await.unwrap_err();
|
|
assert_eq!(err.code, ErrorCode::ArtifactMismatch);
|
|
e.settle("sdk seal cleaned", |s| {
|
|
s.artifacts == 0 && s.store_bytes == 0
|
|
})
|
|
.await;
|
|
}
|
|
|
|
async fn quotas_are_enforced(via: Via) {
|
|
let limits = Limits {
|
|
max_store_bytes: 1000,
|
|
max_artifact_bytes: 600,
|
|
..Limits::default()
|
|
};
|
|
let e = env_with(via, limits, Policy::open()).await;
|
|
let c = e.client("c").await;
|
|
assert_eq!(
|
|
c.artifacts().allocate(601, "x/y").await.unwrap_err().code,
|
|
ErrorCode::QuotaExceeded
|
|
);
|
|
let w = c.artifacts().allocate(600, "x/y").await.unwrap();
|
|
assert_eq!(
|
|
c.artifacts().allocate(401, "x/y").await.unwrap_err().code,
|
|
ErrorCode::QuotaExceeded
|
|
);
|
|
// Sealing copies, so it needs room for both; failing releases the staging object.
|
|
assert_eq!(w.seal().await.unwrap_err().code, ErrorCode::QuotaExceeded);
|
|
e.settle("staging released", |s| s.store_bytes == 0 && s.owners == 0)
|
|
.await;
|
|
|
|
let limits = Limits {
|
|
max_store_bytes: 1200,
|
|
max_artifact_bytes: 600,
|
|
max_owners_per_client: 4,
|
|
reserved_owners_per_client: 1,
|
|
..Limits::default()
|
|
};
|
|
let e = env_with(via, limits, Policy::open()).await;
|
|
let c = e.client("c").await;
|
|
let art = c
|
|
.artifacts()
|
|
.allocate(600, "x/y")
|
|
.await
|
|
.unwrap()
|
|
.seal()
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(e.stats().store_bytes, 600);
|
|
let _w1 = c.artifacts().allocate(1, "x/y").await.unwrap();
|
|
let _w2 = c.artifacts().allocate(1, "x/y").await.unwrap();
|
|
assert_eq!(
|
|
c.artifacts().allocate(1, "x/y").await.unwrap_err().code,
|
|
ErrorCode::QuotaExceeded
|
|
);
|
|
assert_eq!(
|
|
art.retain().await.unwrap_err().code,
|
|
ErrorCode::QuotaExceeded
|
|
);
|
|
}
|
|
|
|
async fn fan_out_shares_one_object_and_the_last_consumer_collects(via: Via) {
|
|
let e = env(via).await;
|
|
let camera = e.client("camera").await;
|
|
camera
|
|
.declare_topic("world.demo.frame", Retained::None)
|
|
.await
|
|
.unwrap();
|
|
let mut viewers = Vec::new();
|
|
for i in 0..3 {
|
|
let c = e.client(&format!("viewer-{i}")).await;
|
|
let s = c
|
|
.subscribe("world.demo.frame", SubscriptionConfig::bounded())
|
|
.await
|
|
.unwrap();
|
|
viewers.push((c, s));
|
|
}
|
|
let frame: Vec<u8> = (0..640 * 480 * 4).map(|i| (i % 256) as u8).collect();
|
|
let art = sealed(&camera, &frame, "image/x-rgba").await;
|
|
let r = camera
|
|
.publish(
|
|
"world.demo.frame",
|
|
obj(json!({"width": 640, "height": 480})),
|
|
&[("frame", &art)],
|
|
)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(r.subscribers, 3);
|
|
drop(art);
|
|
let mut messages = Vec::new();
|
|
for (_, s) in viewers.iter_mut() {
|
|
let m = within("frame", s.next()).await.unwrap();
|
|
assert_eq!(
|
|
m.artifact("frame").unwrap().read_all().await.unwrap(),
|
|
frame
|
|
);
|
|
messages.push(m);
|
|
}
|
|
let s = e
|
|
.settle("three deliveries of one object", |s| s.artifact_roots == 3)
|
|
.await;
|
|
assert_eq!((s.sealed_artifacts, s.store_bytes), (1, frame.len() as u64));
|
|
assert_eq!(e.files("sealed"), 1, "fan-out never copies bytes");
|
|
let path = e.router.store_dir().join("sealed").join(
|
|
&messages[0]
|
|
.artifact("frame")
|
|
.unwrap()
|
|
.reference()
|
|
.artifact_id,
|
|
);
|
|
for (i, m) in messages.drain(..).enumerate() {
|
|
drop(m);
|
|
let left = 2 - i as u64;
|
|
e.settle("one root per consumer", |s| s.artifact_roots == left)
|
|
.await;
|
|
if left > 0 {
|
|
assert!(
|
|
path.exists(),
|
|
"collected while {left} consumers still hold it"
|
|
);
|
|
}
|
|
}
|
|
e.settle_files("sealed", 0).await;
|
|
e.settle("collected", |s| s.artifacts == 0 && s.store_bytes == 0)
|
|
.await;
|
|
}
|
|
|
|
async fn extracted_artifacts_outlive_their_message(via: Via) {
|
|
let e = env(via).await;
|
|
let p = e.client("p").await;
|
|
let r = e.client("r").await;
|
|
p.declare_topic("t.hold", Retained::None).await.unwrap();
|
|
let mut sub = r
|
|
.subscribe("t.hold", SubscriptionConfig::bounded().in_flight(1))
|
|
.await
|
|
.unwrap();
|
|
for i in 0..3 {
|
|
let a = sealed(&p, format!("f{i}").as_bytes(), "text/plain").await;
|
|
p.publish("t.hold", obj(json!({})), &[("f", &a)])
|
|
.await
|
|
.unwrap();
|
|
}
|
|
let m = within("first", sub.next()).await.unwrap();
|
|
let image = m.artifact("f").unwrap();
|
|
drop(m);
|
|
// The extracted handle still owns the delivery, so the credit is not back yet.
|
|
quiet("credit held by the extracted artifact", sub.next()).await;
|
|
assert_eq!(image.read_all().await.unwrap(), b"f0");
|
|
let file = image.open().await.unwrap();
|
|
drop(image);
|
|
quiet("credit held by the open file", sub.next()).await;
|
|
drop(file);
|
|
let m = within("second", sub.next()).await.unwrap();
|
|
// An explicit hold outlives the delivery and does not keep its credit.
|
|
let kept = m.artifact("f").unwrap().retain().await.unwrap();
|
|
drop(m);
|
|
let third = within("third", sub.next()).await.unwrap();
|
|
assert_eq!(kept.read_all().await.unwrap(), b"f1");
|
|
drop((third, kept));
|
|
e.settle("collected", |s| s.artifacts == 0 && s.owners == 0)
|
|
.await;
|
|
}
|
|
|
|
async fn failed_admission_is_atomic(via: Via) {
|
|
let e = env(via).await;
|
|
let reader = e.client("reader").await;
|
|
let server = e.client("server").await;
|
|
let mut raw = e.raw_hello("raw").await;
|
|
assert_eq!(
|
|
code(
|
|
&raw.call(
|
|
"topic.declare",
|
|
json!({"name": "t.atomic", "retained": "latest"})
|
|
)
|
|
.await
|
|
),
|
|
"OK"
|
|
);
|
|
let mut sub = reader
|
|
.subscribe("t.atomic", SubscriptionConfig::bounded())
|
|
.await
|
|
.unwrap();
|
|
let mut svc = server
|
|
.register("example.atomic", ServiceConfig::default())
|
|
.await
|
|
.unwrap();
|
|
let (a, path) = raw.allocate(3).await;
|
|
std::fs::write(&path, b"abc").unwrap();
|
|
let s = raw.seal(&a, Value::Null).await.unwrap();
|
|
let good = |name: &str| json!({"name": name, "ref": s["ref"], "ownerId": s["ownerId"]});
|
|
let before = e.stats();
|
|
let mut wrong_len = s["ref"].clone();
|
|
wrong_len["byteLength"] = json!("4");
|
|
let mut other_store = s["ref"].clone();
|
|
other_store["storeId"] = json!("store-0000000000000000");
|
|
let cases = [
|
|
(
|
|
json!([good("a"), {"name": "b", "ref": s["ref"], "ownerId": "own-7"}]),
|
|
"OWNER_INVALID",
|
|
),
|
|
(
|
|
json!([good("a"), {"name": "b", "ref": wrong_len, "ownerId": s["ownerId"]}]),
|
|
"ARTIFACT_MISMATCH",
|
|
),
|
|
(
|
|
json!([good("a"), {"name": "b", "ref": other_store, "ownerId": s["ownerId"]}]),
|
|
"ARTIFACT_GONE",
|
|
),
|
|
];
|
|
for (i, (atts, want)) in cases.iter().enumerate() {
|
|
let r = raw
|
|
.call_with(
|
|
"publish",
|
|
json!({"topic": "t.atomic", "payload": {}}),
|
|
atts.clone(),
|
|
)
|
|
.await;
|
|
assert_eq!(code(&r), *want);
|
|
let call = json!({"callId": format!("call-{}", i + 1), "target": "example.atomic", "expectedIncarnation": null, "method": "M", "payload": {}});
|
|
let r = raw.call_with("rpc.call", call, atts.clone()).await;
|
|
assert_eq!(
|
|
code(&r),
|
|
*want,
|
|
"semantic refusal advances correlation history but preserves the refusal reason"
|
|
);
|
|
}
|
|
assert_eq!(
|
|
e.stats(),
|
|
before,
|
|
"refused admissions leave no roots, calls or retained values"
|
|
);
|
|
quiet("no partial delivery", sub.next()).await;
|
|
quiet("no partial request", svc.next()).await;
|
|
// Two names for one object: both are delivered, one root is held.
|
|
let r = raw
|
|
.call_with(
|
|
"publish",
|
|
json!({"topic": "t.atomic", "payload": {}}),
|
|
json!([good("a"), good("b")]),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(
|
|
r["topicSequence"], "1",
|
|
"refused publications spent no sequence number"
|
|
);
|
|
let m = within("delivery", sub.next()).await.unwrap();
|
|
assert_eq!(m.attachment_names().collect::<Vec<_>>(), ["a", "b"]);
|
|
assert_eq!(m.artifact("b").unwrap().read_all().await.unwrap(), b"abc");
|
|
// Hold, retained value, delivery: three roots, not four.
|
|
e.settle("one root per holder", |s| s.artifact_roots == 3)
|
|
.await;
|
|
}
|
|
|
|
async fn release_ids_are_watermarked(via: Via) {
|
|
let e = env(via).await;
|
|
let mut raw = e.raw_hello("raw").await;
|
|
let (a1, _) = raw.allocate(4).await;
|
|
let (a2, _) = raw.allocate(4).await;
|
|
let rel = |ids: Value| json!({ "ownerIds": ids });
|
|
assert_eq!(
|
|
code(
|
|
&raw.call("artifact.release", rel(json!([a2["ownerId"], "own-99"])))
|
|
.await
|
|
),
|
|
"OWNER_INVALID"
|
|
);
|
|
assert_eq!(
|
|
e.stats().artifacts,
|
|
2,
|
|
"a batch with a future id releases nothing"
|
|
);
|
|
assert_eq!(
|
|
raw.call("artifact.release", rel(json!([a1["ownerId"]])))
|
|
.await
|
|
.unwrap()["released"],
|
|
"1"
|
|
);
|
|
assert_eq!(
|
|
raw.call("artifact.release", rel(json!([a1["ownerId"]])))
|
|
.await
|
|
.unwrap()["released"],
|
|
"0"
|
|
);
|
|
assert_eq!(
|
|
code(&raw.call("artifact.release", rel(json!(["dlv-1"]))).await),
|
|
"OWNER_INVALID"
|
|
);
|
|
assert_eq!(
|
|
code(&raw.call("artifact.release", rel(json!([]))).await),
|
|
"INVALID_ENVELOPE"
|
|
);
|
|
let many: Vec<String> = (1..=65).map(|i| format!("own-{i}")).collect();
|
|
assert_eq!(
|
|
code(&raw.call("artifact.release", rel(json!(many))).await),
|
|
"INVALID_ENVELOPE"
|
|
);
|
|
let consumed = |ids: Value| json!({ "deliveryIds": ids });
|
|
assert_eq!(
|
|
code(
|
|
&raw.call("delivery.consumed", consumed(json!(["dlv-3"])))
|
|
.await
|
|
),
|
|
"OWNER_INVALID"
|
|
);
|
|
assert_eq!(
|
|
code(
|
|
&raw.call("delivery.consumed", consumed(json!([a2["ownerId"]])))
|
|
.await
|
|
),
|
|
"OWNER_INVALID"
|
|
);
|
|
assert_eq!(
|
|
raw.call(
|
|
"artifact.release",
|
|
rel(json!([a2["ownerId"], a2["ownerId"]]))
|
|
)
|
|
.await
|
|
.unwrap()["released"],
|
|
"1"
|
|
);
|
|
e.settle("all released", |s| s.artifacts == 0 && s.store_bytes == 0)
|
|
.await;
|
|
}
|
|
|
|
async fn owners_are_scoped_to_their_connection(via: Via) {
|
|
let e = env(via).await;
|
|
let owner = e.client("owner").await;
|
|
let art = sealed(&owner, b"private", "text/plain").await;
|
|
let reference = art.reference().to_json();
|
|
let mut thief = e.raw_hello("thief").await;
|
|
let open = |owner_id: &str| json!({"ref": reference, "ownerId": owner_id});
|
|
// Naming the victim's owner id means nothing on another connection.
|
|
assert_eq!(
|
|
code(&thief.call("artifact.open", open(art.owner_id())).await),
|
|
"OWNER_INVALID"
|
|
);
|
|
let (mine, _) = thief.allocate(1).await;
|
|
let my_owner = mine["ownerId"].as_str().unwrap().to_owned();
|
|
assert_eq!(
|
|
code(&thief.call("artifact.open", open(&my_owner)).await),
|
|
"OWNER_INVALID"
|
|
);
|
|
assert_eq!(
|
|
code(&thief.call("artifact.retain", open(&my_owner)).await),
|
|
"OWNER_INVALID"
|
|
);
|
|
assert_eq!(
|
|
code(
|
|
&thief
|
|
.call("artifact.release", json!({"ownerIds": [art.owner_id()]}))
|
|
.await
|
|
),
|
|
"OK"
|
|
);
|
|
assert_eq!(
|
|
art.read_all().await.unwrap(),
|
|
b"private",
|
|
"another connection cannot release it"
|
|
);
|
|
}
|
|
|
|
async fn abandoned_futures_do_not_leak_owners(via: Via) {
|
|
let e = env(via).await;
|
|
let c = e.client("c").await;
|
|
// Each future is polled once (its command is sent) and then dropped.
|
|
let _ = tokio::time::timeout(Duration::ZERO, c.artifacts().allocate(1000, "x/y")).await;
|
|
e.settle("abandoned allocation released", |s| {
|
|
s.artifacts == 0 && s.owners == 0 && s.store_bytes == 0
|
|
})
|
|
.await;
|
|
e.settle_files("staging", 0).await;
|
|
|
|
let mut w = c.artifacts().allocate(4, "x/y").await.unwrap();
|
|
w.write_all(b"seal").unwrap();
|
|
let _ = tokio::time::timeout(Duration::ZERO, w.seal()).await;
|
|
e.settle("abandoned seal released", |s| {
|
|
s.artifacts == 0 && s.owners == 0
|
|
})
|
|
.await;
|
|
|
|
let art = sealed(&c, b"keep", "x/y").await;
|
|
let _ = tokio::time::timeout(Duration::ZERO, art.retain()).await;
|
|
e.settle("abandoned retain released", |s| s.owners == 1)
|
|
.await;
|
|
drop(art);
|
|
e.settle("all released", |s| s.artifacts == 0 && s.owners == 0)
|
|
.await;
|
|
assert_eq!(c.control_errors(), 0);
|
|
}
|
|
|
|
async fn writer_drop_releases_staging(via: Via) {
|
|
let e = env(via).await;
|
|
let c = e.client("c").await;
|
|
let mut w = c.artifacts().allocate(1000, "x/y").await.unwrap();
|
|
w.write_all(&[1; 10]).unwrap();
|
|
assert_eq!(e.files("staging"), 1);
|
|
drop(w);
|
|
e.settle("staging released", |s| {
|
|
s.artifacts == 0 && s.store_bytes == 0
|
|
})
|
|
.await;
|
|
e.settle_files("staging", 0).await;
|
|
}
|
|
|
|
async fn disconnect_releases_all_but_retained(via: Via) {
|
|
let e = env(via).await;
|
|
let producer = e.client("producer").await;
|
|
let reader = e.client("reader").await;
|
|
producer
|
|
.declare_topic("t.keep", Retained::Latest)
|
|
.await
|
|
.unwrap();
|
|
let mut sub = reader
|
|
.subscribe("t.keep", SubscriptionConfig::bounded())
|
|
.await
|
|
.unwrap();
|
|
let a1 = sealed(&producer, b"retained", "x/y").await;
|
|
let a2 = sealed(&producer, b"held", "x/y").await;
|
|
let w3 = producer.artifacts().allocate(10, "x/y").await.unwrap();
|
|
producer
|
|
.publish("t.keep", obj(json!({})), &[("a", &a1)])
|
|
.await
|
|
.unwrap();
|
|
let m = within("delivery", sub.next()).await.unwrap();
|
|
// Close both while handles are still alive; the router releases what they owned.
|
|
producer.close().await;
|
|
reader.close().await;
|
|
let s = e
|
|
.settle("connections gone", |s| s.connections == 0 && s.owners == 0)
|
|
.await;
|
|
assert_eq!(
|
|
(s.artifacts, s.artifact_roots, s.store_bytes),
|
|
(1, 1, 8),
|
|
"only the retained value survives"
|
|
);
|
|
// Handles that outlived their connection are inert.
|
|
assert_eq!(a2.read_all().await.unwrap_err().code, ErrorCode::RouterLost);
|
|
drop((a1, a2, w3, m, sub));
|
|
let admin = e.client("admin").await;
|
|
assert!(admin.clear_topic("t.keep").await.unwrap());
|
|
e.settle("retained released", |s| {
|
|
s.artifacts == 0 && s.store_bytes == 0
|
|
})
|
|
.await;
|
|
}
|
|
|
|
async fn router_restart_invalidates_old_handles(via: Via) {
|
|
let e = env(via).await;
|
|
let c = e.client("c").await;
|
|
c.declare_topic("t.r", Retained::None).await.unwrap();
|
|
let mut sub = c
|
|
.subscribe("t.r", SubscriptionConfig::latest())
|
|
.await
|
|
.unwrap();
|
|
let art = sealed(&c, b"old", "x/y").await;
|
|
e.router.shutdown();
|
|
assert!(within("subscription closed", sub.next()).await.is_none());
|
|
let err = art.read_all().await.unwrap_err();
|
|
assert_eq!(err.code, ErrorCode::RouterLost);
|
|
assert!(c.closed().is_some());
|
|
assert_eq!(
|
|
e.try_client("late").await.unwrap_err().code,
|
|
ErrorCode::RouterLost
|
|
);
|
|
|
|
// A restarted router on the same root is a new store incarnation.
|
|
let mut cfg = RouterConfig::new(e.router.store_root());
|
|
cfg.policy = Policy::open();
|
|
let second = Router::new(cfg).unwrap();
|
|
assert_ne!(second.store_id(), e.router.store_id());
|
|
let mut raw = common::Raw::over(second.connect_in_memory(), second.store_root());
|
|
raw.hello("new").await.unwrap();
|
|
let (mine, _) = raw.allocate(1).await;
|
|
// An old reference is refused even alongside an owner that is live on the new router.
|
|
let r = raw
|
|
.call(
|
|
"artifact.open",
|
|
json!({"ref": art.reference().to_json(), "ownerId": mine["ownerId"]}),
|
|
)
|
|
.await;
|
|
assert_eq!(code(&r), "ARTIFACT_GONE");
|
|
let client = flybus::Client::connect(
|
|
second.connect_in_memory(),
|
|
flybus::ClientConfig::new("sdk", second.store_root()),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
let fresh = sealed(&client, b"new", "x/y").await;
|
|
assert_eq!(fresh.read_all().await.unwrap(), b"new");
|
|
}
|
|
|
|
both_transports!(
|
|
allocate_write_seal_read,
|
|
unsealed_artifacts_cannot_be_used,
|
|
seal_is_immune_to_live_writable_handles,
|
|
seal_checks_length_and_digest,
|
|
quotas_are_enforced,
|
|
fan_out_shares_one_object_and_the_last_consumer_collects,
|
|
extracted_artifacts_outlive_their_message,
|
|
failed_admission_is_atomic,
|
|
release_ids_are_watermarked,
|
|
owners_are_scoped_to_their_connection,
|
|
abandoned_futures_do_not_leak_owners,
|
|
writer_drop_releases_staging,
|
|
disconnect_releases_all_but_retained,
|
|
router_restart_invalidates_old_handles,
|
|
);
|