cf2b8ca4bd
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WV6BzHQiJEvvK9EPc1d4ks
2502 lines
98 KiB
Rust
2502 lines
98 KiB
Rust
//! End-to-end integration over real sockets: discovery → authenticated
|
|
//! selection → paginated browse/search → digest-verified manifest and blob
|
|
//! fetch with Range resume → pinned, budgeted cache → resolver → runtime
|
|
//! states. The fixture serves REAL canonical manifests; every digest the
|
|
//! client checks is a genuine SHA-256 of genuine bytes.
|
|
|
|
// Native socket integration test: std deadlines cannot enter the wasm library.
|
|
#![allow(clippy::disallowed_types, clippy::disallowed_methods)]
|
|
|
|
mod common;
|
|
|
|
use common::{
|
|
payload, response_head, test_root, write_bytes_resp, write_error, write_json_resp, write_raw,
|
|
FixtureOptions, FixtureServer, FixtureStore, ParsedRequest, RawServer,
|
|
};
|
|
use makepad_asset_client::json::{obj, s, Value};
|
|
use makepad_asset_client::{
|
|
AssetClient, CacheBudgets, CatalogQuery, ClientConfig, ClientError, ClientEvent, ClientOutput,
|
|
ClientRequest, ClientRuntime, ClosureBudget, DiscoveryListener, HttpLimits, ResourceSlot,
|
|
RuntimeConfig, SubmitOptions,
|
|
SourceCollectionRegistered, TierPreference,
|
|
};
|
|
use makepad_asset_data::*;
|
|
use std::collections::HashMap;
|
|
use std::net::TcpStream;
|
|
use std::path::{Path, PathBuf};
|
|
use std::str::FromStr;
|
|
use std::sync::{Arc, Mutex};
|
|
|
|
fn now_ms() -> u64 {
|
|
std::time::SystemTime::now()
|
|
.duration_since(std::time::UNIX_EPOCH)
|
|
.unwrap()
|
|
.as_millis() as u64
|
|
}
|
|
|
|
fn fast_limits() -> HttpLimits {
|
|
HttpLimits {
|
|
connect_timeout_ms: 2_000,
|
|
read_timeout_ms: 1_000,
|
|
write_timeout_ms: 2_000,
|
|
head_deadline_ms: 2_000,
|
|
body_deadline_ms: 5_000,
|
|
}
|
|
}
|
|
|
|
fn config(name: &str) -> ClientConfig {
|
|
let mut cfg = ClientConfig::new(test_root(name));
|
|
cfg.http = fast_limits();
|
|
cfg.blob_body_deadline_ms = 5_000;
|
|
cfg
|
|
}
|
|
|
|
fn pin_marker(cache_root: &Path, blob: &BlobId) -> PathBuf {
|
|
cache_root
|
|
.join("pins")
|
|
.join(makepad_asset_client::util::to_hex(blob.as_bytes()))
|
|
}
|
|
|
|
fn wait_runtime(
|
|
runtime: &mut ClientRuntime,
|
|
request: makepad_asset_client::RequestId,
|
|
) -> Result<ClientOutput, ClientError> {
|
|
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
|
|
loop {
|
|
assert!(
|
|
std::time::Instant::now() < deadline,
|
|
"runtime request {request} never finished"
|
|
);
|
|
for event in runtime.poll() {
|
|
match event {
|
|
ClientEvent::Done { id, output } if id == request => return Ok(output),
|
|
ClientEvent::Failed { id, error } if id == request => return Err(error),
|
|
_ => {}
|
|
}
|
|
}
|
|
std::thread::sleep(std::time::Duration::from_millis(5));
|
|
}
|
|
}
|
|
|
|
/// A store with aliased props, a dependency chain, and a game revision.
|
|
fn seeded_store() -> (FixtureStore, Vec<AssetRevisionRef>) {
|
|
let mut store = FixtureStore::default();
|
|
let mut refs = Vec::new();
|
|
// Ten rockets for pagination.
|
|
for i in 0..10u8 {
|
|
let r = store.add_prop(
|
|
10 + i,
|
|
"stock",
|
|
(i == 0).then_some("stock/rocket-launcher"),
|
|
&format!("Rocket {i}"),
|
|
payload(100 + i as u64, 2_000 + i as usize * 100),
|
|
vec![],
|
|
);
|
|
refs.push(r);
|
|
}
|
|
// A crate in another namespace.
|
|
store.add_prop(40, "props", Some("props/crate"), "Wooden Crate", payload(200, 1_500), vec![]);
|
|
(store, refs)
|
|
}
|
|
|
|
/// The VJ filter contract: `exclude_tag` travels on the wire and the SERVER
|
|
/// drops the rows — the client never post-filters a page, so `total`, the
|
|
/// page contents and the cursor walk all agree with the exclusion.
|
|
#[test]
|
|
fn catalog_search_exclude_tag_is_filtered_server_side() {
|
|
let (mut store, refs) = seeded_store();
|
|
// Rockets 1, 3, 5, 7, 9 are intermediates; 3 also carries `keep`.
|
|
for (i, r) in refs.iter().enumerate() {
|
|
if i % 2 == 1 {
|
|
store.tag_asset(r, &["keep", "intermediate"]);
|
|
} else {
|
|
store.tag_asset(r, &["keep"]);
|
|
}
|
|
}
|
|
let fixture = FixtureServer::start(store, FixtureOptions::default());
|
|
let client =
|
|
AssetClient::connect(config("exclude_tag"), fixture.endpoints(), None).unwrap();
|
|
|
|
let mut q = CatalogQuery::text("rocket", 10);
|
|
q.tag = Some("keep".into());
|
|
assert_eq!(client.catalog_search(&q, None).unwrap().total, 10);
|
|
q.exclude_tag = Some("intermediate".into());
|
|
let page = client.catalog_search(&q, None).unwrap();
|
|
assert_eq!(page.total, 5, "the server dropped the intermediates");
|
|
assert_eq!(page.hits.len(), 5);
|
|
assert!(page.next.is_none());
|
|
let kept: Vec<AssetId> =
|
|
refs.iter().step_by(2).map(|r| r.asset_id).collect();
|
|
let mut got: Vec<AssetId> = page.hits.iter().map(|h| h.asset_id).collect();
|
|
got.sort();
|
|
let mut want = kept.clone();
|
|
want.sort();
|
|
assert_eq!(got, want);
|
|
|
|
// Paging over the excluded set: excluded rows interleave the kept ones,
|
|
// and the cursor walk still yields each kept row exactly once.
|
|
q.page_size = 2;
|
|
let mut seen = Vec::new();
|
|
let mut cursor = None;
|
|
for _ in 0..10 {
|
|
let page = client.catalog_search(&q, cursor.as_ref()).unwrap();
|
|
assert_eq!(page.total, 5);
|
|
seen.extend(page.hits.iter().map(|h| h.asset_id));
|
|
match page.next {
|
|
Some(next) => cursor = Some(next),
|
|
None => break,
|
|
}
|
|
}
|
|
seen.sort();
|
|
assert_eq!(seen, want);
|
|
}
|
|
|
|
#[test]
|
|
fn discovery_select_connect_browse_paginate() {
|
|
let token = format!("mpat_{}", "ab".repeat(32));
|
|
let (store, _refs) = seeded_store();
|
|
let options = FixtureOptions {
|
|
server_id: [0x42; 16],
|
|
auth_token: Some(token.clone()),
|
|
..FixtureOptions::default()
|
|
};
|
|
let fixture = FixtureServer::start(store, options);
|
|
|
|
// ---- discovery on a loopback ephemeral port ----
|
|
let listener = DiscoveryListener::start(0, 60_000, now_ms).unwrap();
|
|
fixture.send_beacon(listener.port(), true);
|
|
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
|
|
let candidate = loop {
|
|
if let Some(c) = listener.pick(makepad_asset_client::content_client_caps(), now_ms())
|
|
{
|
|
break c;
|
|
}
|
|
assert!(std::time::Instant::now() < deadline, "no beacon received");
|
|
std::thread::sleep(std::time::Duration::from_millis(20));
|
|
};
|
|
assert_eq!(candidate.server_id, [0x42; 16]);
|
|
assert!(candidate.auth_required);
|
|
|
|
// ---- authenticated selection ----
|
|
let mut cfg = config("discovery_connect");
|
|
cfg.token = Some(token);
|
|
let client = AssetClient::connect_discovered(cfg, &candidate).unwrap();
|
|
assert_eq!(client.server_id(), [0x42; 16]);
|
|
|
|
// ---- catalog search pagination: exact continuation, no dup/skip ----
|
|
let query = CatalogQuery::text("rocket", 3);
|
|
let mut seen = Vec::new();
|
|
let mut cursor = None;
|
|
let mut pages = 0;
|
|
loop {
|
|
let page = client.catalog_search(&query, cursor.as_ref()).unwrap();
|
|
assert_eq!(page.total, 10);
|
|
assert!(page.hits.len() <= 3);
|
|
seen.extend(page.hits.iter().map(|h| h.asset_id));
|
|
pages += 1;
|
|
match page.next {
|
|
Some(next) => cursor = Some(next),
|
|
None => break,
|
|
}
|
|
assert!(pages < 10, "pagination never terminated");
|
|
}
|
|
assert_eq!(pages, 4, "10 hits at page size 3");
|
|
assert_eq!(seen.len(), 10);
|
|
let mut dedup = seen.clone();
|
|
dedup.sort();
|
|
dedup.dedup();
|
|
assert_eq!(dedup.len(), 10, "pagination duplicated a row");
|
|
|
|
// ---- keyset listing with namespace filter ----
|
|
let mut listed = Vec::new();
|
|
let mut cursor = None;
|
|
loop {
|
|
let page = client.assets_page(Some("stock"), cursor.as_ref(), 4).unwrap();
|
|
listed.extend(page.assets.iter().map(|a| a.asset_id));
|
|
for a in &page.assets {
|
|
assert_eq!(a.namespace, "stock");
|
|
}
|
|
match page.next {
|
|
Some(next) => cursor = Some(next),
|
|
None => break,
|
|
}
|
|
}
|
|
assert_eq!(listed.len(), 10);
|
|
let mut sorted = listed.clone();
|
|
sorted.sort_by_key(|id| id.to_string());
|
|
assert_eq!(listed, sorted, "keyset listing must be ordered");
|
|
}
|
|
|
|
#[test]
|
|
fn alias_to_verified_file_end_to_end() {
|
|
let (store, _refs) = seeded_store();
|
|
let glb_bytes = {
|
|
// Original bytes of the aliased rocket's render blob.
|
|
let a = store
|
|
.assets
|
|
.iter()
|
|
.find(|a| a.alias.as_deref() == Some("stock/rocket-launcher"))
|
|
.unwrap();
|
|
let file = &a.manifest.files[0];
|
|
store.blobs[file.blob.as_bytes()].clone()
|
|
};
|
|
let fixture = FixtureServer::start(store, FixtureOptions::default());
|
|
let mut client = AssetClient::connect(config("alias_e2e"), fixture.endpoints(), None).unwrap();
|
|
|
|
// alias → head revision → detail agrees
|
|
let alias = AssetAlias::from_str("stock/rocket-launcher").unwrap();
|
|
let head = client.resolve_alias(&alias).unwrap();
|
|
let detail = client.asset_detail(&head.asset_id).unwrap();
|
|
assert_eq!(detail.latest_published().unwrap().revision, head.head_revision);
|
|
|
|
// manifest: typed, digest-verified, byte-cached
|
|
let manifest = client.fetch_asset_manifest(&head.head_revision).unwrap();
|
|
assert_eq!(manifest.asset_id, head.asset_id);
|
|
|
|
// resolver: role-selected, digest-verified local file
|
|
let resolved = client
|
|
.resolve_file(
|
|
&manifest,
|
|
FileRole::RenderGlb,
|
|
TierPreference::PreferWithAnyFallback(DeviceTier::High),
|
|
7,
|
|
None,
|
|
)
|
|
.unwrap();
|
|
assert_eq!(std::fs::read(resolved.content.as_path().unwrap()).unwrap(), glb_bytes);
|
|
assert_eq!(resolved.byte_len, glb_bytes.len() as u64);
|
|
|
|
// thumbnail is typed and materialized
|
|
let thumb = client.resolve_thumbnail(&manifest).unwrap().expect("mesh thumbnail");
|
|
assert!(thumb.content.as_path().unwrap().exists());
|
|
|
|
// HEAD probe reports the exact size and echoes the identity as a strong
|
|
// ETag (the precondition for trusting a later If-Range resume).
|
|
let head_probe = client.blob_head(&manifest.files[0].blob).unwrap();
|
|
assert_eq!(head_probe.size, manifest.files[0].byte_len);
|
|
assert!(head_probe.etag_matches);
|
|
|
|
// Second resolve is served from cache: zero new blob requests.
|
|
let gets_before = fixture.log.count("GET", "/v1/blobs/");
|
|
let again = client
|
|
.resolve_file(
|
|
&manifest,
|
|
FileRole::RenderGlb,
|
|
TierPreference::PreferWithAnyFallback(DeviceTier::High),
|
|
7,
|
|
None,
|
|
)
|
|
.unwrap();
|
|
assert_eq!(again.content, resolved.content);
|
|
assert_eq!(fixture.log.count("GET", "/v1/blobs/"), gets_before, "cache miss on hot path");
|
|
|
|
// Bytes path agrees with file path.
|
|
let bytes = client
|
|
.resolve_file_bytes(
|
|
&manifest,
|
|
FileRole::RenderGlb,
|
|
TierPreference::PreferWithAnyFallback(DeviceTier::High),
|
|
7,
|
|
)
|
|
.unwrap();
|
|
assert_eq!(bytes, glb_bytes);
|
|
|
|
// Manifest re-fetch is also cache-served.
|
|
let revs_before = fixture.log.count("GET", "/v1/revisions/");
|
|
let _ = client.fetch_asset_manifest(&head.head_revision).unwrap();
|
|
assert_eq!(fixture.log.count("GET", "/v1/revisions/"), revs_before);
|
|
}
|
|
|
|
#[test]
|
|
fn range_resume_after_connection_kill() {
|
|
let mut store = FixtureStore::default();
|
|
let big = payload(555, 300_000);
|
|
let r = store.add_prop(50, "stock", None, "Big Rocket", big.clone(), vec![]);
|
|
let fixture = FixtureServer::start(store, FixtureOptions::default());
|
|
let mut client = AssetClient::connect(config("resume"), fixture.endpoints(), None).unwrap();
|
|
|
|
let manifest = client.fetch_asset_manifest(&r.revision).unwrap();
|
|
let file = manifest.files[0].clone();
|
|
|
|
// First attempt dies after ~90KB of body.
|
|
*fixture.knobs.kill_blob_after.lock().unwrap() = Some(90_000);
|
|
|
|
let mut progress_points = Vec::new();
|
|
let mut progress = |bytes: u64, total: u64| progress_points.push((bytes, total));
|
|
let path = client
|
|
.fetch_blob(&file.blob, Some(file.byte_len), Some(&mut progress))
|
|
.unwrap();
|
|
assert_eq!(std::fs::read(&path).unwrap(), big);
|
|
|
|
// Exactly two GETs: the killed one and the resume.
|
|
assert_eq!(fixture.log.count("GET", "/v1/blobs/"), 2);
|
|
let resume = fixture.log.last_matching("GET", "/v1/blobs/").unwrap();
|
|
let range = resume.header("range").expect("resume must use Range");
|
|
let start: u64 = range
|
|
.strip_prefix("bytes=")
|
|
.and_then(|r| r.strip_suffix('-'))
|
|
.and_then(|r| r.parse().ok())
|
|
.expect("open-ended range shape");
|
|
assert!(start > 0 && start < big.len() as u64, "resume offset {start}");
|
|
let if_range = resume.header("if-range").expect("resume must gate on If-Range");
|
|
assert_eq!(if_range, format!("\"{}\"", file.blob));
|
|
|
|
// Progress is monotonic and ends complete.
|
|
assert!(progress_points.windows(2).all(|w| w[0].0 <= w[1].0));
|
|
assert_eq!(progress_points.last().unwrap(), &(big.len() as u64, big.len() as u64));
|
|
}
|
|
|
|
#[test]
|
|
fn restart_resumes_partial_across_client_instances() {
|
|
let mut store = FixtureStore::default();
|
|
let big = payload(556, 200_000);
|
|
let r = store.add_prop(51, "stock", None, "Interrupted Rocket", big.clone(), vec![]);
|
|
let fixture = FixtureServer::start(store, FixtureOptions::default());
|
|
|
|
let root = test_root("restart_resume");
|
|
let file = {
|
|
let mut cfg = config("unused");
|
|
cfg.cache_root = root.clone();
|
|
cfg.max_transfer_attempts = 1; // fail hard on the kill
|
|
let mut client = AssetClient::connect(cfg, fixture.endpoints(), None).unwrap();
|
|
let manifest = client.fetch_asset_manifest(&r.revision).unwrap();
|
|
let file = manifest.files[0].clone();
|
|
*fixture.knobs.kill_blob_after.lock().unwrap() = Some(60_000);
|
|
let err = client.fetch_blob(&file.blob, Some(file.byte_len), None).unwrap_err();
|
|
assert!(matches!(err, ClientError::Io { .. }), "{err:?}");
|
|
file
|
|
// client drops; partial stays on disk
|
|
};
|
|
|
|
// A NEW client on the same cache root resumes instead of restarting.
|
|
let mut cfg = config("unused2");
|
|
cfg.cache_root = root;
|
|
let mut client = AssetClient::connect(cfg, fixture.endpoints(), None).unwrap();
|
|
let path = client.fetch_blob(&file.blob, Some(file.byte_len), None).unwrap();
|
|
assert_eq!(std::fs::read(&path).unwrap(), big);
|
|
let resume = fixture.log.last_matching("GET", "/v1/blobs/").unwrap();
|
|
assert!(resume.header("range").is_some(), "second process must resume, not restart");
|
|
}
|
|
|
|
#[test]
|
|
fn offline_resolution_from_cache_and_honest_miss() {
|
|
let (store, refs) = seeded_store();
|
|
let fixture = FixtureServer::start(store, FixtureOptions::default());
|
|
let mut client = AssetClient::connect(config("offline"), fixture.endpoints(), None).unwrap();
|
|
|
|
let manifest = client.fetch_asset_manifest(&refs[0].revision).unwrap();
|
|
let resolved = client
|
|
.resolve_file(
|
|
&manifest,
|
|
FileRole::RenderGlb,
|
|
TierPreference::PreferWithAnyFallback(DeviceTier::Low),
|
|
7,
|
|
None,
|
|
)
|
|
.unwrap();
|
|
|
|
// Server goes away entirely.
|
|
let mut fixture = fixture;
|
|
fixture.control.stop();
|
|
fixture.data.stop();
|
|
|
|
// Cached content still resolves, fully verified, with zero network.
|
|
let again = client
|
|
.resolve_file(
|
|
&manifest,
|
|
FileRole::RenderGlb,
|
|
TierPreference::PreferWithAnyFallback(DeviceTier::Low),
|
|
7,
|
|
None,
|
|
)
|
|
.unwrap();
|
|
assert_eq!(again.content, resolved.content);
|
|
let cached = client.cached_blob(&manifest.files[0].blob).unwrap();
|
|
assert!(cached.is_some());
|
|
|
|
// Uncached content fails with an explicit transport error — no guess, no
|
|
// fallback, no stale substitute.
|
|
let other = client.fetch_asset_manifest(&refs[5].revision);
|
|
match other {
|
|
Err(ClientError::Io { .. }) | Err(ClientError::Timeout { .. }) => {}
|
|
other => panic!("offline miss must be an explicit transport error: {other:?}"),
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn pinning_survives_eviction_pressure_end_to_end() {
|
|
let mut store = FixtureStore::default();
|
|
let precious = payload(700, 30_000);
|
|
let precious_ref = store.add_prop(60, "stock", None, "Precious", precious.clone(), vec![]);
|
|
for i in 0..4u8 {
|
|
store.add_prop(61 + i, "stock", None, &format!("Filler {i}"), payload(710 + i as u64, 40_000), vec![]);
|
|
}
|
|
let fixture = FixtureServer::start(store, FixtureOptions::default());
|
|
|
|
let mut cfg = config("pin_pressure");
|
|
cfg.cache = CacheBudgets {
|
|
// Room for ~two big blobs (plus manifests/thumbs): heavy pressure.
|
|
max_total_bytes: 100_000,
|
|
max_object_bytes: 60_000,
|
|
max_partial_bytes: 100_000,
|
|
stale_partial_ms: 1_000_000,
|
|
max_ram_bytes: 512 * 1024,
|
|
};
|
|
let mut client = AssetClient::connect(cfg, fixture.endpoints(), None).unwrap();
|
|
|
|
let manifest = client.fetch_asset_manifest(&precious_ref.revision).unwrap();
|
|
let blob = manifest.files[0].blob;
|
|
client.pin_blob(&blob).unwrap();
|
|
client.fetch_blob(&blob, Some(manifest.files[0].byte_len), None).unwrap();
|
|
let gets_for_precious = fixture.log.count("GET", &format!("/v1/blobs/{blob}"));
|
|
|
|
// Churn: pull every filler through the small cache.
|
|
for i in 0..4u8 {
|
|
let asset_id = makepad_asset_data::AssetId::from_bytes([61 + i; 16]);
|
|
let detail = client.asset_detail(&asset_id).unwrap();
|
|
let rev = detail.latest_published().unwrap().revision;
|
|
let m = client.fetch_asset_manifest(&rev).unwrap();
|
|
client.fetch_blob(&m.files[0].blob, Some(m.files[0].byte_len), None).unwrap();
|
|
}
|
|
assert!(client.cache_stats().evictions > 0, "pressure never evicted anything");
|
|
|
|
// The pinned blob is still local — no refetch happened.
|
|
assert!(client.cached_blob(&blob).unwrap().is_some(), "pinned blob was evicted");
|
|
assert_eq!(
|
|
fixture.log.count("GET", &format!("/v1/blobs/{blob}")),
|
|
gets_for_precious,
|
|
"pinned blob was refetched"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn dependency_closure_bounded_and_verified() {
|
|
let mut store = FixtureStore::default();
|
|
let leaf = store.add_prop(72, "stock", None, "Leaf", payload(900, 800), vec![]);
|
|
let mid = store.add_prop(71, "stock", None, "Mid", payload(901, 800), vec![leaf]);
|
|
let root_ref = store.add_prop(70, "stock", None, "Root", payload(902, 800), vec![mid]);
|
|
let fixture = FixtureServer::start(store, FixtureOptions::default());
|
|
let mut client = AssetClient::connect(config("closure"), fixture.endpoints(), None).unwrap();
|
|
|
|
let closure = client.resolve_closure(&root_ref, ClosureBudget::default()).unwrap();
|
|
let ids: Vec<_> = closure.iter().map(|(r, _)| r.asset_id).collect();
|
|
assert_eq!(ids.len(), 3);
|
|
assert_eq!(closure[0].0, root_ref, "BFS order starts at the root");
|
|
// Every manifest's declared pair was proven.
|
|
for (r, m) in &closure {
|
|
assert_eq!(m.asset_id, r.asset_id);
|
|
}
|
|
|
|
// Asset budget refusal.
|
|
let err = client
|
|
.resolve_closure(&root_ref, ClosureBudget { max_assets: 2, max_depth: 8 })
|
|
.unwrap_err();
|
|
assert!(matches!(err, ClientError::OverBudget { what: "closure assets", .. }), "{err:?}");
|
|
|
|
// Depth budget refusal.
|
|
let err = client
|
|
.resolve_closure(&root_ref, ClosureBudget { max_assets: 10, max_depth: 1 })
|
|
.unwrap_err();
|
|
assert!(matches!(err, ClientError::OverBudget { what: "closure depth", .. }), "{err:?}");
|
|
}
|
|
|
|
/// The bridge for tools that take a FILE rather than bytes (an AO bake, a
|
|
/// rig pass, an OS drag-out): a verified on-disk path for catalog content.
|
|
/// It stays thin-client-legal because the object is named by its digest and
|
|
/// re-hashed before the path is handed out — a materialisation of the
|
|
/// revision, never a second source of truth.
|
|
#[test]
|
|
fn blob_path_materialises_verified_content_and_re_fetches_a_corrupted_object() {
|
|
let mut store = FixtureStore::default();
|
|
let payload = vec![7u8; 60_000];
|
|
let blob = store.add_blob(payload.clone());
|
|
let other = store.add_blob(b"a second, different object".to_vec());
|
|
let fixture = FixtureServer::start(store, FixtureOptions::default());
|
|
let mut client = AssetClient::connect(config("blob_path"), fixture.endpoints(), None).unwrap();
|
|
|
|
let path = client.blob_path(&blob, Some(payload.len() as u64)).unwrap();
|
|
assert!(path.is_file(), "the object is on disk at {}", path.display());
|
|
assert_eq!(std::fs::read(&path).unwrap(), payload, "and it is the real payload");
|
|
// Digest-keyed: the same blob resolves to the same path, and a second
|
|
// call is served from the cache rather than the network.
|
|
assert_eq!(client.blob_path(&blob, None).unwrap(), path);
|
|
assert_ne!(client.blob_path(&other, None).unwrap(), path, "distinct objects, distinct paths");
|
|
|
|
// A path is only handed out for bytes that still hash to the digest: a
|
|
// corrupted object is removed and re-fetched, never returned.
|
|
std::fs::write(&path, b"tampered").unwrap();
|
|
let again = client.blob_path(&blob, Some(payload.len() as u64)).unwrap();
|
|
assert_eq!(std::fs::read(&again).unwrap(), payload, "corruption re-fetched, not served");
|
|
|
|
// Asking for a path does not spend the RAM budget on a file the caller
|
|
// is about to read from disk.
|
|
client.clear_ram_cache();
|
|
let cold = client.blob_path(&blob, None).unwrap();
|
|
assert_eq!(std::fs::read(cold).unwrap(), payload);
|
|
assert_eq!(client.ram_cache_bytes().0, 0, "a path fetch stays out of RAM");
|
|
}
|
|
|
|
#[test]
|
|
fn ram_cache_evicts_under_its_budget_and_refetches_verified_after_forget() {
|
|
// Five blobs, a budget that fits two: the client must stay inside it
|
|
// and still answer every fetch with verified bytes.
|
|
let mut store = FixtureStore::default();
|
|
let payloads: Vec<Vec<u8>> = (0..5u8).map(|i| vec![i + 1; 40_000]).collect();
|
|
let blobs: Vec<BlobId> = payloads
|
|
.iter()
|
|
.map(|bytes| store.add_blob(bytes.clone()))
|
|
.collect();
|
|
let fixture = FixtureServer::start(store, FixtureOptions::default());
|
|
let mut cfg = config("ram_budget");
|
|
cfg.cache.max_ram_bytes = 100_000;
|
|
let mut client = AssetClient::connect(cfg, fixture.endpoints(), None).unwrap();
|
|
|
|
for (blob, expect) in blobs.iter().zip(&payloads) {
|
|
let got = client.fetch_blob_bytes(blob, Some(expect.len() as u64)).unwrap();
|
|
assert_eq!(&got, expect, "every fetch is the real, verified payload");
|
|
let (used, budget) = client.ram_cache_bytes();
|
|
assert!(used <= budget, "ram cache blew its budget: {used} > {budget}");
|
|
}
|
|
let (used, budget) = client.ram_cache_bytes();
|
|
assert_eq!(budget, 100_000);
|
|
assert!(used <= budget && used > 0, "some residency, under budget: {used}");
|
|
|
|
// Forget drops residency; the next fetch re-materialises and re-verifies
|
|
// from the server (or the disk cache) rather than serving a ghost.
|
|
let hot = &blobs[4];
|
|
assert!(client.forget_blob(hot), "the newest fetch was resident");
|
|
let (after_forget, _) = client.ram_cache_bytes();
|
|
assert!(after_forget < used, "forget freed its bytes: {after_forget} !< {used}");
|
|
let again = client.fetch_blob_bytes(hot, Some(payloads[4].len() as u64)).unwrap();
|
|
assert_eq!(&again, &payloads[4], "re-fetch is verified, not a ghost");
|
|
|
|
// Clearing empties it without breaking any later fetch.
|
|
client.clear_ram_cache();
|
|
assert_eq!(client.ram_cache_bytes().0, 0);
|
|
let cold = client.fetch_blob_bytes(&blobs[0], Some(payloads[0].len() as u64)).unwrap();
|
|
assert_eq!(&cold, &payloads[0]);
|
|
}
|
|
|
|
#[test]
|
|
fn ram_cache_budget_holds_while_lanes_fetch_together() {
|
|
// Lane clones share one RAM cache. Eight threads pulling the same six
|
|
// blobs must never push it past the budget.
|
|
let mut store = FixtureStore::default();
|
|
let payloads: Vec<Vec<u8>> = (0..6u8).map(|i| vec![i + 9; 30_000]).collect();
|
|
let blobs: Vec<BlobId> = payloads
|
|
.iter()
|
|
.map(|bytes| store.add_blob(bytes.clone()))
|
|
.collect();
|
|
let fixture = FixtureServer::start(store, FixtureOptions::default());
|
|
let mut cfg = config("ram_lanes");
|
|
cfg.cache.max_ram_bytes = 90_000;
|
|
let client = AssetClient::connect(cfg, fixture.endpoints(), None).unwrap();
|
|
|
|
std::thread::scope(|scope| {
|
|
for lane in 0..8 {
|
|
let mut lane_client = client.lane_clone();
|
|
let blobs = &blobs;
|
|
let payloads = &payloads;
|
|
scope.spawn(move || {
|
|
for round in 0..6 {
|
|
let i = (lane + round) % blobs.len();
|
|
let got = lane_client
|
|
.fetch_blob_bytes(&blobs[i], Some(payloads[i].len() as u64))
|
|
.unwrap();
|
|
assert_eq!(got, payloads[i]);
|
|
let (used, budget) = lane_client.ram_cache_bytes();
|
|
assert!(used <= budget, "lane saw {used} > {budget}");
|
|
}
|
|
});
|
|
}
|
|
});
|
|
let (used, budget) = client.ram_cache_bytes();
|
|
assert!(used <= budget, "after the lanes: {used} > {budget}");
|
|
}
|
|
|
|
#[test]
|
|
fn game_manifest_and_blob_fetch() {
|
|
let mut store = FixtureStore::default();
|
|
let game_rev = store.add_game(90, "Fixture Kart");
|
|
let splash_len = {
|
|
let bytes = store.game_manifests[game_rev.as_bytes()].clone();
|
|
let m = makepad_asset_data::GameRevisionManifest::from_canonical_bytes(&bytes).unwrap();
|
|
m.splash_byte_len
|
|
};
|
|
let fixture = FixtureServer::start(store, FixtureOptions::default());
|
|
let mut client = AssetClient::connect(config("game"), fixture.endpoints(), None).unwrap();
|
|
|
|
let game = client.fetch_game_manifest(&game_rev).unwrap();
|
|
assert_eq!(game.name, "Fixture Kart");
|
|
// The splash blob materializes with its declared length verified.
|
|
let splash = client.fetch_blob_bytes(&game.splash_blob, Some(game.splash_byte_len)).unwrap();
|
|
assert_eq!(splash.len() as u64, splash_len);
|
|
// Lock blob has no declared length in the manifest: fetched under the
|
|
// budget cap, still digest-verified.
|
|
let lock = client.fetch_blob_bytes(&game.lock_blob, None).unwrap();
|
|
assert!(!lock.is_empty());
|
|
// Cached game manifest round-trips.
|
|
let again = client.fetch_game_manifest(&game_rev).unwrap();
|
|
assert_eq!(again, game);
|
|
}
|
|
|
|
#[test]
|
|
fn complete_partial_with_unknown_length_commits_via_416() {
|
|
// A fully downloaded partial whose expected length is unknown: the resume
|
|
// request lands at the server's size, gets 416, and the client PROVES the
|
|
// partial by digest locally instead of refetching anything.
|
|
let mut store = FixtureStore::default();
|
|
let bytes = payload(660, 50_000);
|
|
let r = store.add_prop(96, "stock", None, "Complete Partial", bytes.clone(), vec![]);
|
|
let blob = store.assets.iter().find(|a| a.asset_id == r.asset_id).unwrap().manifest.files[0]
|
|
.blob;
|
|
let fixture = FixtureServer::start(store, FixtureOptions::default());
|
|
|
|
let root = test_root("complete_partial");
|
|
{
|
|
let mut cache = makepad_asset_client::ContentCache::open(
|
|
&root,
|
|
CacheBudgets::default_v1(),
|
|
now_ms(),
|
|
)
|
|
.unwrap();
|
|
let mut w = cache.open_partial(blob.as_bytes()).unwrap();
|
|
w.write(&bytes).unwrap();
|
|
}
|
|
|
|
let mut cfg = config("unused3");
|
|
cfg.cache_root = root;
|
|
let mut client = AssetClient::connect(cfg, fixture.endpoints(), None).unwrap();
|
|
let path = client.fetch_blob(&blob, None, None).unwrap();
|
|
assert_eq!(std::fs::read(&path).unwrap(), bytes);
|
|
// One request (the 416 probe), zero body bytes refetched.
|
|
assert_eq!(fixture.log.count("GET", "/v1/blobs/"), 1);
|
|
let probe = fixture.log.last_matching("GET", "/v1/blobs/").unwrap();
|
|
assert_eq!(
|
|
probe.header("range").unwrap(),
|
|
format!("bytes={}-", bytes.len()),
|
|
"probe must resume at the partial's end"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn runtime_states_are_explicit() {
|
|
let (store, refs) = seeded_store();
|
|
let fixture = FixtureServer::start(store, FixtureOptions::default());
|
|
let client = AssetClient::connect(config("runtime"), fixture.endpoints(), None).unwrap();
|
|
let mut runtime = ClientRuntime::start(client).unwrap();
|
|
|
|
// A search slot and a failing alias slot.
|
|
let search_id = runtime
|
|
.submit(ClientRequest::CatalogSearch {
|
|
query: CatalogQuery::text("rocket", 5),
|
|
cursor: None,
|
|
})
|
|
.unwrap();
|
|
let missing_alias = AssetAlias::from_str("stock/does-not-exist").unwrap();
|
|
let fail_id = runtime.submit(ClientRequest::ResolveAlias { alias: missing_alias }).unwrap();
|
|
let manifest_id = runtime
|
|
.submit(ClientRequest::FetchAssetManifest { rev: refs[0].revision })
|
|
.unwrap();
|
|
|
|
let mut search_slot: ResourceSlot<u64> = ResourceSlot::default();
|
|
search_slot.begin(search_id);
|
|
assert!(search_slot.state.is_loading());
|
|
|
|
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
|
|
let mut fail_seen = false;
|
|
let mut manifest_seen = false;
|
|
let mut started_order = Vec::new();
|
|
while !(search_slot.state.is_ready() && fail_seen && manifest_seen) {
|
|
assert!(std::time::Instant::now() < deadline, "runtime never finished");
|
|
for event in runtime.poll() {
|
|
if let ClientEvent::Started { id } = &event {
|
|
started_order.push(*id);
|
|
}
|
|
let consumed = search_slot.on_event(&event, |out| match out {
|
|
makepad_asset_client::ClientOutput::CatalogPage(p) => Some(p.total),
|
|
_ => None,
|
|
});
|
|
if consumed {
|
|
continue;
|
|
}
|
|
match event {
|
|
ClientEvent::Failed { id, error } if id == fail_id => {
|
|
assert!(matches!(error, ClientError::NotFound { .. }), "{error:?}");
|
|
fail_seen = true;
|
|
}
|
|
ClientEvent::Done { id, .. } if id == manifest_id => {
|
|
manifest_seen = true;
|
|
}
|
|
_ => {}
|
|
}
|
|
}
|
|
std::thread::sleep(std::time::Duration::from_millis(10));
|
|
}
|
|
assert_eq!(search_slot.state.ready(), Some(&10u64));
|
|
// Lanes run requests in parallel, so completion order across requests is
|
|
// NOT a guarantee any more (tests/runtime_lanes.rs pins what is): every
|
|
// submitted request starts exactly once, and its own events stay ordered.
|
|
started_order.sort_unstable();
|
|
assert_eq!(started_order, vec![search_id, fail_id, manifest_id]);
|
|
|
|
runtime.shutdown();
|
|
}
|
|
|
|
#[test]
|
|
fn runtime_reports_blob_progress() {
|
|
let mut store = FixtureStore::default();
|
|
let big = payload(999, 1_500_000);
|
|
let r = store.add_prop(95, "stock", None, "Huge", big.clone(), vec![]);
|
|
let fixture = FixtureServer::start(store, FixtureOptions::default());
|
|
let mut client = AssetClient::connect(config("runtime_prog"), fixture.endpoints(), None).unwrap();
|
|
let manifest = client.fetch_asset_manifest(&r.revision).unwrap();
|
|
let file = manifest.files[0].clone();
|
|
let mut runtime = ClientRuntime::start(client).unwrap();
|
|
|
|
let id = runtime
|
|
.submit(ClientRequest::FetchBlob {
|
|
blob: file.blob,
|
|
expected_len: Some(file.byte_len),
|
|
pin: true,
|
|
})
|
|
.unwrap();
|
|
let mut progress = Vec::new();
|
|
let mut done_path = None;
|
|
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(15);
|
|
while done_path.is_none() {
|
|
assert!(std::time::Instant::now() < deadline, "blob fetch never finished");
|
|
for event in runtime.poll() {
|
|
match event {
|
|
ClientEvent::Progress { id: eid, bytes, total } if eid == id => {
|
|
progress.push((bytes, total));
|
|
}
|
|
ClientEvent::Done { id: eid, output } if eid == id => match output {
|
|
makepad_asset_client::ClientOutput::Blob { content, .. } => {
|
|
done_path = Some(content.as_path().unwrap().to_path_buf());
|
|
}
|
|
other => panic!("wrong output: {other:?}"),
|
|
},
|
|
ClientEvent::Failed { id: eid, error } if eid == id => {
|
|
panic!("blob fetch failed: {error}");
|
|
}
|
|
_ => {}
|
|
}
|
|
}
|
|
std::thread::sleep(std::time::Duration::from_millis(5));
|
|
}
|
|
assert_eq!(std::fs::read(done_path.unwrap()).unwrap(), big);
|
|
assert!(progress.len() >= 2, "throttled progress still reports interior points");
|
|
assert!(progress.windows(2).all(|w| w[0].0 <= w[1].0), "progress must be monotonic");
|
|
assert_eq!(progress.last().unwrap(), &(big.len() as u64, big.len() as u64));
|
|
runtime.shutdown();
|
|
}
|
|
|
|
#[test]
|
|
fn runtime_blob_pin_is_transactional_and_async_unpin_is_idempotent() {
|
|
let bytes = payload(1001, 48_000);
|
|
let mut store = FixtureStore::default();
|
|
let blob = store.add_blob(bytes.clone());
|
|
let fixture = FixtureServer::start(store, FixtureOptions::default());
|
|
let cfg = config("runtime_blob_lease");
|
|
let cache_root = cfg.cache_root.clone();
|
|
let client = AssetClient::connect(cfg, fixture.endpoints(), None).unwrap();
|
|
let mut runtime = ClientRuntime::start(client).unwrap();
|
|
let marker = pin_marker(&cache_root, &blob);
|
|
|
|
// A refusal must not leave the old pre-fetch pin marker behind.
|
|
let failed = runtime
|
|
.submit(ClientRequest::FetchBlob {
|
|
blob,
|
|
expected_len: Some(bytes.len() as u64 + 1),
|
|
pin: true,
|
|
})
|
|
.unwrap();
|
|
assert!(
|
|
matches!(wait_runtime(&mut runtime, failed), Err(ClientError::SizeMismatch { .. })),
|
|
"wrong declared length must refuse"
|
|
);
|
|
assert!(!marker.exists(), "failed fetch leaked a durable pin marker");
|
|
|
|
let fetched = runtime
|
|
.submit(ClientRequest::FetchBlob {
|
|
blob,
|
|
expected_len: Some(bytes.len() as u64),
|
|
pin: true,
|
|
})
|
|
.unwrap();
|
|
match wait_runtime(&mut runtime, fetched).expect("verified fetch") {
|
|
ClientOutput::Blob { blob: got, content } => {
|
|
assert_eq!(got, blob);
|
|
assert_eq!(std::fs::read(content.as_path().unwrap()).unwrap(), bytes);
|
|
}
|
|
other => panic!("wrong fetch output: {other:?}"),
|
|
}
|
|
assert!(marker.exists(), "successful pinned fetch did not pin its object");
|
|
|
|
for _ in 0..2 {
|
|
let unpin = runtime.submit(ClientRequest::UnpinBlob { blob }).unwrap();
|
|
match wait_runtime(&mut runtime, unpin).expect("async unpin") {
|
|
ClientOutput::BlobUnpinned { blob: got } => assert_eq!(got, blob),
|
|
other => panic!("wrong unpin output: {other:?}"),
|
|
}
|
|
assert!(!marker.exists(), "unpin left a durable pin marker");
|
|
}
|
|
runtime.shutdown();
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// artifact publication (write path)
|
|
// ---------------------------------------------------------------------------
|
|
|
|
#[test]
|
|
fn publish_artifact_roundtrips_manifest_blobs_alias_and_annotation() {
|
|
use makepad_asset_client::{PublishFile, PublishRequest, PublishThumbnail};
|
|
use makepad_asset_data::{AssetKind, MediaType, ThumbnailMedia};
|
|
|
|
let token = format!("mpat_{}", "5c".repeat(32));
|
|
let fx = FixtureServer::start(
|
|
FixtureStore::default(),
|
|
FixtureOptions { auth_token: Some(token.clone()), ..FixtureOptions::default() },
|
|
);
|
|
let mut cfg = config("publish_rt");
|
|
cfg.token = Some(token);
|
|
let mut client = AssetClient::connect(cfg, fx.endpoints(), None).expect("connect");
|
|
|
|
let artifact = payload(31, 5_000);
|
|
let thumb = payload(32, 1_200);
|
|
let mut request = PublishRequest::new(
|
|
"gen",
|
|
AssetKind::Video,
|
|
"Neon drift",
|
|
PublishFile {
|
|
bytes: artifact.clone(),
|
|
media: MediaType::Mp4,
|
|
role: FileRole::Video,
|
|
media_millis: 5_200,
|
|
dims: None,
|
|
},
|
|
PublishThumbnail {
|
|
bytes: thumb.clone(),
|
|
media: ThumbnailMedia::Png,
|
|
width: 512,
|
|
height: 512,
|
|
views: Vec::new(),
|
|
},
|
|
);
|
|
request.alias = Some(AssetAlias::from_str("gen/neon-drift").unwrap());
|
|
request.categories = vec!["music-video".into()];
|
|
request.prompt = "a neon drift over wet asphalt".into();
|
|
|
|
let published = client.publish_artifact(&request).expect("publish");
|
|
assert_eq!(published.alias.as_ref().unwrap().as_str(), "gen/neon-drift");
|
|
|
|
// The published revision fetches back, digest-verified + decoded, with
|
|
// the playable file and the mandatory thumbnail intact.
|
|
let manifest = client.fetch_asset_manifest(&published.revision).expect("manifest");
|
|
assert_eq!(manifest.asset_id, published.asset_id);
|
|
assert_eq!(manifest.kind, AssetKind::Video);
|
|
assert_eq!(manifest.files[0].blob, published.artifact_blob);
|
|
assert_eq!(manifest.thumbnail.as_ref().unwrap().blob, published.thumbnail_blob);
|
|
|
|
// The artifact bytes round-trip through the verified cache path.
|
|
let bytes = client
|
|
.fetch_blob_bytes(&published.artifact_blob, Some(artifact.len() as u64))
|
|
.expect("blob");
|
|
assert_eq!(bytes, artifact);
|
|
|
|
// The alias resolves to the published head; the detail reports it as the
|
|
// latest published candidate.
|
|
let alias = client.resolve_alias(request.alias.as_ref().unwrap()).expect("alias");
|
|
assert_eq!(alias.asset_id, published.asset_id);
|
|
assert_eq!(alias.head_revision, published.revision);
|
|
let detail = client.asset_detail(&published.asset_id).expect("detail");
|
|
assert_eq!(detail.latest_published().unwrap().revision, published.revision);
|
|
|
|
// The annotation landed (title recorded by the fixture's 204 route),
|
|
// and it landed BEFORE the publish (kind-stamped publish events).
|
|
assert_eq!(
|
|
fx.published.lock().unwrap().annotations.get(&published.asset_id.to_string()),
|
|
Some(&"Neon drift".to_string())
|
|
);
|
|
let log = fx.log.requests.lock().unwrap();
|
|
let idx_of = |m: &str, frag: &str| {
|
|
log.iter()
|
|
.position(|r| r.method == m && r.target.contains(frag))
|
|
.unwrap_or(usize::MAX)
|
|
};
|
|
assert!(
|
|
idx_of("PUT", "/annotation") < idx_of("POST", "/publish"),
|
|
"annotation must precede publish so the publish event carries the kind"
|
|
);
|
|
drop(log);
|
|
|
|
// Re-publish onto the SAME asset id (register 409 path): new revision,
|
|
// alias head moves.
|
|
let mut again = request.clone();
|
|
again.asset_id = Some(published.asset_id);
|
|
again.artifact.bytes[0] ^= 0xff;
|
|
let second = client.publish_artifact(&again).expect("re-publish");
|
|
assert_eq!(second.asset_id, published.asset_id);
|
|
assert_ne!(second.revision, published.revision);
|
|
let alias = client.resolve_alias(request.alias.as_ref().unwrap()).expect("alias 2");
|
|
assert_eq!(alias.head_revision, second.revision);
|
|
let detail = client.asset_detail(&published.asset_id).expect("detail 2");
|
|
assert_eq!(detail.latest_published().unwrap().revision, second.revision);
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// runtime cancellation, cache-root locking, session, concurrent discovery
|
|
// ---------------------------------------------------------------------------
|
|
|
|
#[test]
|
|
fn runtime_cancel_skips_queued_and_aborts_in_flight() {
|
|
let mut store = FixtureStore::default();
|
|
// A blob large enough to drip for a while: 512KB in 32KB/25ms chunks
|
|
// (~400ms total) — cancellation lands mid-transfer deterministically.
|
|
let big = payload(700, 512 * 1024);
|
|
let big_id = store.add_blob(big.clone());
|
|
let small = payload(701, 20_000);
|
|
let small_id = store.add_blob(small.clone());
|
|
let fx = FixtureServer::start(store, FixtureOptions::default());
|
|
*fx.knobs.drip_blob.lock().unwrap() = Some((32 * 1024, 25));
|
|
|
|
let cfg = config("cancel_rt");
|
|
let cache_root = cfg.cache_root.clone();
|
|
let client = AssetClient::connect(cfg, fx.endpoints(), None).unwrap();
|
|
// This test is about the QUEUE: one worker per lane and both fetches
|
|
// pinned to the same lane, so the second is provably still queued when
|
|
// it is cancelled. (With the default pool it would simply run in
|
|
// parallel — which is the point of the lanes, proven in
|
|
// tests/runtime_lanes.rs.)
|
|
let mut runtime = ClientRuntime::start_with(
|
|
client,
|
|
RuntimeConfig { fast_workers: 1, bulk_workers: 1, ..RuntimeConfig::default_v1() },
|
|
)
|
|
.unwrap();
|
|
let id_big = runtime
|
|
.submit_with(
|
|
ClientRequest::FetchBlob {
|
|
blob: big_id,
|
|
expected_len: Some(big.len() as u64),
|
|
pin: true,
|
|
},
|
|
SubmitOptions::bulk(),
|
|
)
|
|
.unwrap();
|
|
let id_queued = runtime
|
|
.submit_with(
|
|
ClientRequest::FetchBlob {
|
|
blob: small_id,
|
|
expected_len: Some(small.len() as u64),
|
|
pin: true,
|
|
},
|
|
SubmitOptions::bulk(),
|
|
)
|
|
.unwrap();
|
|
// Give the worker time to start the drip transfer, then cancel BOTH:
|
|
// the in-flight one aborts between chunks, the queued one never starts.
|
|
std::thread::sleep(std::time::Duration::from_millis(120));
|
|
runtime.cancel(id_big);
|
|
runtime.cancel(id_queued);
|
|
|
|
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
|
|
let mut failures = std::collections::HashMap::new();
|
|
while failures.len() < 2 && std::time::Instant::now() < deadline {
|
|
for event in runtime.poll() {
|
|
if let ClientEvent::Failed { id, error } = event {
|
|
failures.insert(id, error);
|
|
}
|
|
}
|
|
std::thread::sleep(std::time::Duration::from_millis(10));
|
|
}
|
|
assert_eq!(
|
|
failures.get(&id_big),
|
|
Some(&ClientError::Cancelled),
|
|
"in-flight transfer must abort as Cancelled"
|
|
);
|
|
assert_eq!(
|
|
failures.get(&id_queued),
|
|
Some(&ClientError::Cancelled),
|
|
"queued request must be skipped as Cancelled"
|
|
);
|
|
assert!(
|
|
!pin_marker(&cache_root, &big_id).exists(),
|
|
"cancelled in-flight fetch leaked a durable pin marker"
|
|
);
|
|
assert!(
|
|
!pin_marker(&cache_root, &small_id).exists(),
|
|
"cancelled queued fetch leaked a durable pin marker"
|
|
);
|
|
|
|
// The aborted partial stays resumable: a fresh fetch (drip off)
|
|
// completes and verifies.
|
|
*fx.knobs.drip_blob.lock().unwrap() = None;
|
|
let id_retry = runtime
|
|
.submit(ClientRequest::FetchBlob {
|
|
blob: big_id,
|
|
expected_len: Some(big.len() as u64),
|
|
pin: false,
|
|
})
|
|
.unwrap();
|
|
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
|
|
let mut done = false;
|
|
while !done && std::time::Instant::now() < deadline {
|
|
for event in runtime.poll() {
|
|
match event {
|
|
ClientEvent::Done { id, .. } if id == id_retry => done = true,
|
|
ClientEvent::Failed { id, error } if id == id_retry => {
|
|
panic!("retry after cancel failed: {error}")
|
|
}
|
|
_ => {}
|
|
}
|
|
}
|
|
std::thread::sleep(std::time::Duration::from_millis(10));
|
|
}
|
|
assert!(done, "cancelled partial must resume cleanly");
|
|
}
|
|
|
|
#[test]
|
|
fn cache_root_is_process_exclusive_and_restart_clean() {
|
|
use makepad_asset_client::{CacheBudgets, ContentCache};
|
|
let root = test_root("cache_lock");
|
|
let first = ContentCache::open(&root, CacheBudgets::default_v1(), 1).expect("first open");
|
|
// A second live cache on the SAME root refuses — the single-owner
|
|
// contract is enforced, not just documented.
|
|
match ContentCache::open(&root, CacheBudgets::default_v1(), 2) {
|
|
Err(ClientError::CacheBusy) => {}
|
|
Err(other) => panic!("second owner must refuse with CacheBusy, got {other:?}"),
|
|
Ok(_) => panic!("second owner must refuse with CacheBusy, got Ok"),
|
|
}
|
|
// Releasing the first owner (clean or crashed process — the OS drops
|
|
// the lock either way) makes reopening immediate.
|
|
drop(first);
|
|
let again = ContentCache::open(&root, CacheBudgets::default_v1(), 3).expect("reopen");
|
|
drop(again);
|
|
}
|
|
|
|
#[test]
|
|
fn session_connector_hands_over_and_stops_cleanly() {
|
|
use makepad_asset_client::{SessionConfig, SessionConnector, SessionMsg, SessionStatus};
|
|
|
|
let fx = FixtureServer::start(FixtureStore::default(), FixtureOptions::default());
|
|
let mut cfg = SessionConfig::new(test_root("session_up"));
|
|
cfg.endpoints = Some(fx.endpoints());
|
|
cfg.media_lanes = vec!["video-a".into(), "video-b".into(), "audio".into()];
|
|
cfg.retry_min_ms = 100;
|
|
cfg.retry_max_ms = 200;
|
|
let mut connector = SessionConnector::start(cfg).expect("start");
|
|
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
|
|
let mut handles = None;
|
|
while handles.is_none() && std::time::Instant::now() < deadline {
|
|
for msg in connector.poll() {
|
|
if let SessionMsg::Up(up) = msg {
|
|
handles = Some(*up);
|
|
}
|
|
}
|
|
std::thread::sleep(std::time::Duration::from_millis(20));
|
|
}
|
|
let handles = handles.expect("session must come up against the fixture");
|
|
assert_eq!(handles.media.len(), 3, "one runtime per configured lane");
|
|
assert_eq!(handles.server_id, fx.options.server_id);
|
|
handles.shutdown();
|
|
|
|
// Wrong pinned identity: the connector reports honest retries and stop()
|
|
// ends it promptly.
|
|
let mut cfg = SessionConfig::new(test_root("session_pin"));
|
|
cfg.endpoints = Some(fx.endpoints());
|
|
cfg.server_id = Some([0x99; 16]);
|
|
cfg.retry_min_ms = 100;
|
|
cfg.retry_max_ms = 200;
|
|
let mut connector = SessionConnector::start(cfg).expect("start 2");
|
|
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
|
|
let mut retried = false;
|
|
while !retried && std::time::Instant::now() < deadline {
|
|
for msg in connector.poll() {
|
|
if let SessionMsg::Status(SessionStatus::Retrying { .. }) = msg {
|
|
retried = true;
|
|
}
|
|
}
|
|
std::thread::sleep(std::time::Duration::from_millis(20));
|
|
}
|
|
assert!(retried, "identity mismatch must surface as Retrying");
|
|
connector.stop();
|
|
|
|
// Config validation refuses duplicate/overlapping lanes.
|
|
let mut bad = SessionConfig::new(test_root("session_bad"));
|
|
bad.media_lanes = vec!["a".into(), "a".into()];
|
|
assert!(SessionConnector::start(bad).is_err());
|
|
}
|
|
|
|
#[cfg(unix)]
|
|
#[test]
|
|
fn discovery_listeners_share_the_port_on_one_host() {
|
|
// The regression this pins: two apps on one machine must BOTH be able
|
|
// to bind the discovery port (SO_REUSEADDR + SO_REUSEPORT). Beacons are
|
|
// broadcast on real LANs and reach every group member; loopback unicast
|
|
// delivery lands on at least one listener, which is asserted loosely.
|
|
let a = DiscoveryListener::start(0, 5_000, now_ms).expect("first listener");
|
|
let port = a.port();
|
|
let b = DiscoveryListener::start(port, 5_000, now_ms).expect(
|
|
"second listener on the SAME port must bind (reuse group)",
|
|
);
|
|
let fx = FixtureServer::start(FixtureStore::default(), FixtureOptions::default());
|
|
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
|
|
let mut seen = false;
|
|
while !seen && std::time::Instant::now() < deadline {
|
|
fx.send_beacon(port, false);
|
|
std::thread::sleep(std::time::Duration::from_millis(50));
|
|
let now = now_ms();
|
|
seen = !a.snapshot(now).is_empty() || !b.snapshot(now).is_empty();
|
|
}
|
|
assert!(seen, "a beacon must reach the reuse group");
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// import + immutable derived-variant routes
|
|
//
|
|
// Real-process coverage of these routes belongs in the Asset Server crate
|
|
// (`libs/asset/store` e2e). This suite stays hermetic: a local TCP fixture
|
|
// that speaks the wire contract. A clean-checkout `cargo test` must not
|
|
// require a prebuilt `makepad-asset-store` binary.
|
|
// ---------------------------------------------------------------------------
|
|
|
|
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
|
enum ImportTamper {
|
|
None,
|
|
SourceDigest,
|
|
ImportRevision,
|
|
ImportGetId,
|
|
ImportEntryRevision,
|
|
VariantBytes,
|
|
VariantEtag,
|
|
VariantSetId,
|
|
ResolveDigest,
|
|
SourcePageCursor,
|
|
SourcePageOrder,
|
|
}
|
|
|
|
struct ImportPlaneState {
|
|
write_token: String,
|
|
read_token: String,
|
|
server_id: [u8; 16],
|
|
sources: HashMap<String, (SourceCollectionId, Vec<u8>)>,
|
|
imports: HashMap<[u8; 32], (Vec<u8>, ImportReportDtoStored)>,
|
|
variants: HashMap<[u8; 32], Vec<u8>>,
|
|
sets: HashMap<[u8; 32], Vec<u8>>,
|
|
tamper: ImportTamper,
|
|
}
|
|
|
|
struct ImportReportDtoStored {
|
|
entries: Vec<(PackEntryKey, AssetId, AssetRevisionId, AssetAlias)>,
|
|
}
|
|
|
|
struct ImportPlane {
|
|
control: RawServer,
|
|
data: RawServer,
|
|
state: Arc<Mutex<ImportPlaneState>>,
|
|
write_token: String,
|
|
read_token: String,
|
|
}
|
|
|
|
impl ImportPlane {
|
|
fn start() -> ImportPlane {
|
|
Self::start_with_id([0x51; 16])
|
|
}
|
|
|
|
fn start_with_id(server_id: [u8; 16]) -> ImportPlane {
|
|
let write_token = format!("mpat_{}", "aa".repeat(32));
|
|
let read_token = format!("mpat_{}", "bb".repeat(32));
|
|
let state = Arc::new(Mutex::new(ImportPlaneState {
|
|
write_token: write_token.clone(),
|
|
read_token: read_token.clone(),
|
|
server_id,
|
|
sources: HashMap::new(),
|
|
imports: HashMap::new(),
|
|
variants: HashMap::new(),
|
|
sets: HashMap::new(),
|
|
tamper: ImportTamper::None,
|
|
}));
|
|
let control = {
|
|
let state = state.clone();
|
|
RawServer::start(Arc::new(move |req, stream| {
|
|
import_control(&state, &req, stream);
|
|
}))
|
|
};
|
|
let data = RawServer::start(Arc::new(|_req, stream| {
|
|
write_error(stream, 404, "not found");
|
|
}));
|
|
ImportPlane { control, data, state, write_token, read_token }
|
|
}
|
|
|
|
fn endpoints(&self) -> makepad_asset_client::ApiEndpoints {
|
|
makepad_asset_client::ApiEndpoints {
|
|
control: self.control.addr,
|
|
data: self.data.addr,
|
|
}
|
|
}
|
|
|
|
fn set_tamper(&self, tamper: ImportTamper) {
|
|
self.state.lock().unwrap().tamper = tamper;
|
|
}
|
|
|
|
fn seed_variant(&self, bytes: Vec<u8>) -> DerivedVariantId {
|
|
let id = DerivedVariantId::hash_of(&bytes);
|
|
self.state.lock().unwrap().variants.insert(*id.as_bytes(), bytes);
|
|
id
|
|
}
|
|
|
|
fn seed_source(&self, collection: SourceCollection) {
|
|
let bytes = collection.to_canonical_bytes().unwrap();
|
|
let digest = SourceCollectionId::hash_of(&bytes);
|
|
self.state
|
|
.lock()
|
|
.unwrap()
|
|
.sources
|
|
.insert(collection.id, (digest, bytes));
|
|
}
|
|
}
|
|
|
|
fn bearer_of(req: &ParsedRequest) -> Option<&str> {
|
|
req.header("authorization")
|
|
.and_then(|v| v.strip_prefix("Bearer "))
|
|
}
|
|
|
|
fn import_auth(state: &ImportPlaneState, req: &ParsedRequest) -> Result<bool, u16> {
|
|
// `true` = write principal, `false` = authenticated reader.
|
|
match bearer_of(req) {
|
|
Some(t) if t == state.write_token => Ok(true),
|
|
Some(t) if t == state.read_token => Ok(false),
|
|
_ => Err(401),
|
|
}
|
|
}
|
|
|
|
fn import_control(state: &Mutex<ImportPlaneState>, req: &ParsedRequest, stream: &mut TcpStream) {
|
|
let segs = req.segs();
|
|
let seg = |i: usize| segs.get(i).map(String::as_str).unwrap_or("");
|
|
if req.method == "GET" && seg(0) == "v1" && seg(1) == "health" {
|
|
let server_id = state.lock().unwrap().server_id;
|
|
write_json_resp(
|
|
stream,
|
|
200,
|
|
&obj(vec![
|
|
("server_id", s(common::hex(&server_id))),
|
|
("protocol_version", Value::Int(1)),
|
|
]),
|
|
);
|
|
return;
|
|
}
|
|
let write = {
|
|
let st = state.lock().unwrap();
|
|
match import_auth(&st, req) {
|
|
Ok(w) => w,
|
|
Err(401) => {
|
|
write_error(stream, 401, "unauthenticated");
|
|
return;
|
|
}
|
|
Err(_) => unreachable!(),
|
|
}
|
|
};
|
|
match (req.method.as_str(), seg(0), seg(1), segs.len()) {
|
|
("GET", "v1", "assets", 2) => {
|
|
write_json_resp(
|
|
stream,
|
|
200,
|
|
&obj(vec![("assets", Value::Arr(Vec::new())), ("cursor", Value::Null)]),
|
|
);
|
|
}
|
|
("PUT", "v1", "import-sources", 2) => {
|
|
if !write {
|
|
write_json_resp(
|
|
stream,
|
|
403,
|
|
&obj(vec![("error", s("denied")), ("capability", s("import_source"))]),
|
|
);
|
|
return;
|
|
}
|
|
let Ok(collection) = SourceCollection::from_canonical_bytes(&req.body) else {
|
|
write_error(stream, 400, "malformed source collection");
|
|
return;
|
|
};
|
|
let digest = SourceCollectionId::hash_of(&req.body);
|
|
let mut st = state.lock().unwrap();
|
|
if let Some((existing, _)) = st.sources.get(&collection.id) {
|
|
if *existing != digest {
|
|
write_error(stream, 409, "conflict");
|
|
return;
|
|
}
|
|
} else {
|
|
st.sources
|
|
.insert(collection.id.clone(), (digest, req.body.clone()));
|
|
}
|
|
let echoed = if st.tamper == ImportTamper::SourceDigest {
|
|
SourceCollectionId::from_bytes({
|
|
let mut b = *digest.as_bytes();
|
|
b[31] ^= 0xff;
|
|
b
|
|
})
|
|
} else {
|
|
digest
|
|
};
|
|
write_json_resp(
|
|
stream,
|
|
201,
|
|
&obj(vec![
|
|
("source_id", s(collection.id)),
|
|
("digest", s(echoed.to_string())),
|
|
]),
|
|
);
|
|
}
|
|
("GET", "v1", "import-sources", 2) => {
|
|
let explicit =
|
|
req.query_get("limit").is_some() || req.query_get("cursor").is_some();
|
|
let limit = if explicit {
|
|
match req.query_get("limit") {
|
|
None => 100usize,
|
|
Some(t) => {
|
|
if t.is_empty()
|
|
|| t.len() > 6
|
|
|| !t.bytes().all(|b| b.is_ascii_digit())
|
|
{
|
|
write_error(stream, 400, "malformed limit");
|
|
return;
|
|
}
|
|
let n: u64 = match t.parse() {
|
|
Ok(n) if n > 0 => n,
|
|
_ => {
|
|
write_error(stream, 400, "malformed limit");
|
|
return;
|
|
}
|
|
};
|
|
n.min(500) as usize
|
|
}
|
|
}
|
|
} else {
|
|
512
|
|
};
|
|
if let Some(c) = req.query_get("cursor") {
|
|
if !makepad_asset_client::wire::source_cursor_ok(&c) {
|
|
write_error(stream, 400, "malformed source cursor");
|
|
return;
|
|
}
|
|
}
|
|
let after = req.query_get("cursor");
|
|
let st = state.lock().unwrap();
|
|
let mut ids: Vec<_> = st.sources.keys().cloned().collect();
|
|
ids.sort();
|
|
if st.tamper == ImportTamper::SourcePageOrder {
|
|
ids.reverse();
|
|
}
|
|
let mut rows = Vec::new();
|
|
let mut last_source_id = None;
|
|
let mut more = false;
|
|
for id in ids {
|
|
if after.as_ref().is_some_and(|cursor| id.as_str() <= cursor.as_str()) {
|
|
continue;
|
|
}
|
|
if rows.len() == limit {
|
|
more = true;
|
|
break;
|
|
}
|
|
last_source_id = Some(id.clone());
|
|
let (digest, bytes) = st.sources.get(&id).unwrap();
|
|
let c = SourceCollection::from_canonical_bytes(bytes).unwrap();
|
|
rows.push(obj(vec![
|
|
("source_id", s(id)),
|
|
("title", s(c.title)),
|
|
("license", s(c.terms.license)),
|
|
("credits", s(c.terms.credits)),
|
|
("digest", s(digest.to_string())),
|
|
]));
|
|
}
|
|
if more && !explicit {
|
|
write_error(stream, 413, "source listing requires pagination");
|
|
return;
|
|
}
|
|
let cursor = match (more, last_source_id, st.tamper) {
|
|
(true, Some(id), ImportTamper::SourcePageCursor) => s(format!("{id}-x")),
|
|
(true, Some(id), _) => s(id),
|
|
_ => Value::Null,
|
|
};
|
|
write_json_resp(
|
|
stream,
|
|
200,
|
|
&obj(vec![("sources", Value::Arr(rows)), ("cursor", cursor)]),
|
|
);
|
|
}
|
|
("POST", "v1", "imports", 2) => {
|
|
if !write {
|
|
write_json_resp(
|
|
stream,
|
|
403,
|
|
&obj(vec![("error", s("denied")), ("capability", s("import_run"))]),
|
|
);
|
|
return;
|
|
}
|
|
let Ok(manifest) = ImportManifest::from_canonical_bytes(&req.body) else {
|
|
write_error(stream, 400, "malformed import");
|
|
return;
|
|
};
|
|
let irev = ImportRevisionId::hash_of(&req.body);
|
|
let mut st = state.lock().unwrap();
|
|
let (created, entries) = if let Some((_, stored)) = st.imports.get(irev.as_bytes()) {
|
|
(false, stored.entries.clone())
|
|
} else {
|
|
let Some((registered, registered_bytes)) = st.sources.get(&manifest.source_id) else {
|
|
write_error(stream, 404, "not found");
|
|
return;
|
|
};
|
|
if *registered != manifest.source_collection {
|
|
write_error(stream, 409, "conflict");
|
|
return;
|
|
}
|
|
let registered = SourceCollection::from_canonical_bytes(registered_bytes).unwrap();
|
|
if manifest.rights != registered.terms {
|
|
write_error(stream, 409, "conflict");
|
|
return;
|
|
}
|
|
let mut entries = Vec::new();
|
|
for asset in &manifest.assets {
|
|
let produced = manifest.asset_manifest_for(asset, &irev).unwrap();
|
|
let revision = produced.revision().unwrap();
|
|
entries.push((
|
|
asset.key.clone(),
|
|
produced.asset_id,
|
|
revision,
|
|
manifest.alias_for(&asset.key).unwrap(),
|
|
));
|
|
}
|
|
st.imports.insert(
|
|
*irev.as_bytes(),
|
|
(
|
|
req.body.clone(),
|
|
ImportReportDtoStored { entries: entries.clone() },
|
|
),
|
|
);
|
|
(true, entries)
|
|
};
|
|
let echoed = if st.tamper == ImportTamper::ImportRevision {
|
|
ImportRevisionId::from_bytes({
|
|
let mut b = *irev.as_bytes();
|
|
b[0] ^= 0xff;
|
|
b
|
|
})
|
|
} else {
|
|
irev
|
|
};
|
|
let rows: Vec<Value> = entries
|
|
.into_iter()
|
|
.map(|(key, asset_id, revision, alias)| {
|
|
let revision = if st.tamper == ImportTamper::ImportEntryRevision {
|
|
AssetRevisionId::from_bytes({
|
|
let mut b = *revision.as_bytes();
|
|
b[0] ^= 0xff;
|
|
b
|
|
})
|
|
} else {
|
|
revision
|
|
};
|
|
obj(vec![
|
|
("key", s(key.as_str().to_string())),
|
|
("asset_id", s(asset_id.to_string())),
|
|
("revision", s(revision.to_string())),
|
|
("alias", s(alias.as_str().to_string())),
|
|
])
|
|
})
|
|
.collect();
|
|
write_json_resp(
|
|
stream,
|
|
if created { 201 } else { 200 },
|
|
&obj(vec![
|
|
("import_revision", s(echoed.to_string())),
|
|
("created", Value::Bool(created)),
|
|
("entries", Value::Arr(rows)),
|
|
]),
|
|
);
|
|
}
|
|
("GET", "v1", "imports", 3) => {
|
|
let Ok(irev) = seg(2).parse::<ImportRevisionId>() else {
|
|
write_error(stream, 400, "malformed import revision");
|
|
return;
|
|
};
|
|
let st = state.lock().unwrap();
|
|
let Some((bytes, stored)) = st.imports.get(irev.as_bytes()) else {
|
|
write_error(stream, 404, "not found");
|
|
return;
|
|
};
|
|
let manifest = ImportManifest::from_canonical_bytes(bytes).unwrap();
|
|
let echoed = if st.tamper == ImportTamper::ImportGetId {
|
|
ImportRevisionId::from_bytes([0xff; 32])
|
|
} else {
|
|
irev
|
|
};
|
|
let rows: Vec<Value> = stored
|
|
.entries
|
|
.iter()
|
|
.map(|(key, asset_id, revision, alias)| {
|
|
obj(vec![
|
|
("key", s(key.as_str().to_string())),
|
|
("asset_id", s(asset_id.to_string())),
|
|
("revision", s(revision.to_string())),
|
|
("alias", s(alias.as_str().to_string())),
|
|
])
|
|
})
|
|
.collect();
|
|
write_json_resp(
|
|
stream,
|
|
200,
|
|
&obj(vec![
|
|
("import_revision", s(echoed.to_string())),
|
|
("source_id", s(manifest.source_id)),
|
|
("pack_name", s(manifest.pack_name)),
|
|
("pack_version", s(manifest.pack_version)),
|
|
("license", s(manifest.rights.license)),
|
|
("credits", s(manifest.rights.credits)),
|
|
("entries", Value::Arr(rows)),
|
|
]),
|
|
);
|
|
}
|
|
("GET", "v1", "derived-variants", 3) => {
|
|
let Ok(id) = seg(2).parse::<DerivedVariantId>() else {
|
|
write_error(stream, 400, "malformed derived variant");
|
|
return;
|
|
};
|
|
let st = state.lock().unwrap();
|
|
let Some(mut bytes) = st.variants.get(id.as_bytes()).cloned() else {
|
|
write_error(stream, 404, "not found");
|
|
return;
|
|
};
|
|
if st.tamper == ImportTamper::VariantBytes {
|
|
bytes[0] ^= 0xff;
|
|
}
|
|
let etag = if st.tamper == ImportTamper::VariantEtag {
|
|
format!("\"{}\"", DerivedVariantId::from_bytes([0x00; 32]))
|
|
} else {
|
|
format!("\"{id}\"")
|
|
};
|
|
write_bytes_resp(
|
|
stream,
|
|
200,
|
|
"application/octet-stream",
|
|
&bytes,
|
|
&[("ETag", &etag), ("Cache-Control", "private, max-age=31536000, immutable")],
|
|
);
|
|
}
|
|
("POST", "v1", "variant-sets", 2) => {
|
|
if !write {
|
|
write_json_resp(
|
|
stream,
|
|
403,
|
|
&obj(vec![("error", s("denied")), ("capability", s("asset_publish"))]),
|
|
);
|
|
return;
|
|
}
|
|
let Ok(body) = makepad_asset_client::json::parse(&req.body) else {
|
|
write_error(stream, 400, "malformed json");
|
|
return;
|
|
};
|
|
let Some(base_asset) = body
|
|
.get("base_asset")
|
|
.and_then(Value::as_str)
|
|
.and_then(|s| s.parse::<AssetId>().ok())
|
|
else {
|
|
write_error(stream, 400, "malformed asset id");
|
|
return;
|
|
};
|
|
let Some(base_revision) = body
|
|
.get("base_revision")
|
|
.and_then(Value::as_str)
|
|
.and_then(|s| s.parse::<AssetRevisionId>().ok())
|
|
else {
|
|
write_error(stream, 400, "malformed asset revision");
|
|
return;
|
|
};
|
|
let Some(arr) = body.get("variants").and_then(Value::as_arr) else {
|
|
write_error(stream, 400, "variants must be an array");
|
|
return;
|
|
};
|
|
let mut variants = Vec::new();
|
|
for v in arr {
|
|
let Some(id) = v.as_str().and_then(|t| t.parse::<DerivedVariantId>().ok()) else {
|
|
write_error(stream, 400, "malformed derived variant");
|
|
return;
|
|
};
|
|
variants.push(id);
|
|
}
|
|
let base = AssetRevisionRef { asset_id: base_asset, revision: base_revision };
|
|
let mut set = VariantSetManifest {
|
|
base,
|
|
variants: variants.clone(),
|
|
policy_version: RESOLUTION_POLICY_V1,
|
|
};
|
|
set.canonicalize();
|
|
let Ok(set_bytes) = set.to_canonical_bytes() else {
|
|
write_error(stream, 400, "invalid variant set");
|
|
return;
|
|
};
|
|
let set_id = VariantSetId::hash_of(&set_bytes);
|
|
let mut st = state.lock().unwrap();
|
|
for v in &set.variants {
|
|
let Some(bytes) = st.variants.get(v.as_bytes()) else {
|
|
write_error(stream, 404, "not found");
|
|
return;
|
|
};
|
|
let Ok(manifest) = DerivedVariantManifest::from_canonical_bytes(bytes) else {
|
|
write_error(stream, 400, "malformed derived variant");
|
|
return;
|
|
};
|
|
if manifest.base != base {
|
|
write_error(stream, 409, "conflict");
|
|
return;
|
|
}
|
|
}
|
|
st.sets.insert(*set_id.as_bytes(), set_bytes);
|
|
let echoed = if st.tamper == ImportTamper::VariantSetId {
|
|
VariantSetId::from_bytes({
|
|
let mut b = *set_id.as_bytes();
|
|
b[0] ^= 0xff;
|
|
b
|
|
})
|
|
} else {
|
|
set_id
|
|
};
|
|
write_json_resp(
|
|
stream,
|
|
201,
|
|
&obj(vec![("variant_set", s(echoed.to_string()))]),
|
|
);
|
|
}
|
|
("GET", "v1", "variant-sets", 3) => {
|
|
let Ok(id) = seg(2).parse::<VariantSetId>() else {
|
|
write_error(stream, 400, "malformed variant set");
|
|
return;
|
|
};
|
|
let st = state.lock().unwrap();
|
|
let Some(bytes) = st.sets.get(id.as_bytes()) else {
|
|
write_error(stream, 404, "not found");
|
|
return;
|
|
};
|
|
let etag = format!("\"{id}\"");
|
|
write_bytes_resp(
|
|
stream,
|
|
200,
|
|
"application/octet-stream",
|
|
bytes,
|
|
&[("ETag", &etag), ("Cache-Control", "private, max-age=31536000, immutable")],
|
|
);
|
|
}
|
|
("POST", "v1", "variant-resolutions", 2) => {
|
|
let Ok(body) = makepad_asset_client::json::parse(&req.body) else {
|
|
write_error(stream, 400, "malformed json");
|
|
return;
|
|
};
|
|
let Some(set_id) = body
|
|
.get("variant_set")
|
|
.and_then(Value::as_str)
|
|
.and_then(|s| s.parse::<VariantSetId>().ok())
|
|
else {
|
|
write_error(stream, 400, "malformed variant set");
|
|
return;
|
|
};
|
|
let Some(p) = body.get("profile") else {
|
|
write_error(stream, 400, "missing profile");
|
|
return;
|
|
};
|
|
let Some(tier) = p
|
|
.get("tier")
|
|
.and_then(Value::as_str)
|
|
.and_then(parse_tier_name)
|
|
else {
|
|
write_error(stream, 400, "malformed profile tier");
|
|
return;
|
|
};
|
|
let Some(accept) = p.get("accept").and_then(Value::as_arr) else {
|
|
write_error(stream, 400, "profile accept must be an array");
|
|
return;
|
|
};
|
|
let mut profile = ClientProfile {
|
|
policy_version: p
|
|
.get("policy_version")
|
|
.and_then(Value::as_u64)
|
|
.unwrap_or(RESOLUTION_POLICY_V1 as u64) as u32,
|
|
tier,
|
|
max_texture_dim: p.get("max_texture_dim").and_then(Value::as_u64).unwrap_or(0) as u32,
|
|
max_triangles: p.get("max_triangles").and_then(Value::as_u64).unwrap_or(0) as u32,
|
|
max_variant_bytes: p.get("max_variant_bytes").and_then(Value::as_u64).unwrap_or(0),
|
|
accept_png: false,
|
|
accept_jpeg: false,
|
|
accept_glb: false,
|
|
accept_bin: false,
|
|
};
|
|
for a in accept {
|
|
match a.as_str() {
|
|
Some("png") => profile.accept_png = true,
|
|
Some("jpeg") => profile.accept_jpeg = true,
|
|
Some("glb") => profile.accept_glb = true,
|
|
Some("bin") => profile.accept_bin = true,
|
|
_ => {
|
|
write_error(stream, 400, "malformed profile accept entry");
|
|
return;
|
|
}
|
|
}
|
|
}
|
|
let st = state.lock().unwrap();
|
|
let Some(set_bytes) = st.sets.get(set_id.as_bytes()) else {
|
|
write_error(stream, 404, "not found");
|
|
return;
|
|
};
|
|
let set = VariantSetManifest::from_canonical_bytes(set_bytes).unwrap();
|
|
let mut variants = Vec::new();
|
|
for v in &set.variants {
|
|
let Some(bytes) = st.variants.get(v.as_bytes()) else {
|
|
write_error(stream, 404, "not found");
|
|
return;
|
|
};
|
|
variants.push(DerivedVariantManifest::from_canonical_bytes(bytes).unwrap());
|
|
}
|
|
let map = resolve_variants(&set, &variants, &profile).unwrap();
|
|
let digest = if st.tamper == ImportTamper::ResolveDigest {
|
|
ResolvedMapDigest::from_bytes({
|
|
let mut b = *map.digest().unwrap().as_bytes();
|
|
b[0] ^= 0xff;
|
|
b
|
|
})
|
|
} else {
|
|
map.digest().unwrap()
|
|
};
|
|
let entries: Vec<Value> = map
|
|
.entries
|
|
.iter()
|
|
.map(|e| {
|
|
obj(vec![
|
|
(
|
|
"role",
|
|
s(makepad_asset_client::dto::variant_role_name(e.role)),
|
|
),
|
|
("variant", s(e.variant.to_string())),
|
|
(
|
|
"blobs",
|
|
Value::Arr(e.blobs.iter().map(|b| s(b.to_string())).collect()),
|
|
),
|
|
])
|
|
})
|
|
.collect();
|
|
write_json_resp(
|
|
stream,
|
|
200,
|
|
&obj(vec![
|
|
("digest", s(digest.to_string())),
|
|
("variant_set", s(map.set.to_string())),
|
|
("profile", s(map.profile.to_string())),
|
|
("entries", Value::Arr(entries)),
|
|
]),
|
|
);
|
|
}
|
|
_ => write_error(stream, 404, "not found"),
|
|
}
|
|
}
|
|
|
|
fn parse_tier_name(s: &str) -> Option<DeviceTier> {
|
|
Some(match s {
|
|
"any" => DeviceTier::Any,
|
|
"low" => DeviceTier::Low,
|
|
"medium" => DeviceTier::Medium,
|
|
"high" => DeviceTier::High,
|
|
_ => return None,
|
|
})
|
|
}
|
|
|
|
fn fixture_rights() -> Rights {
|
|
Rights {
|
|
license: "CC0-1.0".into(),
|
|
license_revision: String::new(),
|
|
terms_digest: Some(sha256(b"CC0-1.0 legal text")),
|
|
terms_url: "https://creativecommons.org/publicdomain/zero/1.0/".into(),
|
|
credits: "Kenney (kenney.nl)".into(),
|
|
source: "https://kenney.nl/assets/space-kit".into(),
|
|
source_archive: Some(sha256(b"space-kit-1.0.zip")),
|
|
redistribution: Redistribution::Allowed,
|
|
derivatives: DerivativePolicy::Allowed,
|
|
}
|
|
}
|
|
|
|
fn fixture_collection() -> SourceCollection {
|
|
SourceCollection {
|
|
id: "kenney".into(),
|
|
title: "Kenney game assets".into(),
|
|
origin: SourceOrigin::Upload,
|
|
terms: fixture_rights(),
|
|
}
|
|
}
|
|
|
|
fn fixture_import() -> ImportManifest {
|
|
let glb = b"watchtower-glb";
|
|
let preview = b"watchtower-png";
|
|
let mut manifest = ImportManifest {
|
|
source_collection: fixture_collection().digest().unwrap(),
|
|
source_id: "kenney".into(),
|
|
pack_name: "space-kit".into(),
|
|
pack_version: "1.0".into(),
|
|
policy_version: IMPORT_ASSET_ID_POLICY_V1,
|
|
assets: vec![ImportAsset {
|
|
key: "models/watchtower".parse().unwrap(),
|
|
kind: AssetKind::Prop,
|
|
files: vec![ImportFile {
|
|
path: "models/watchtower.glb".into(),
|
|
file: AssetFile {
|
|
role: FileRole::RenderGlb,
|
|
tier: DeviceTier::Any,
|
|
lod: 0,
|
|
media: MediaType::Glb,
|
|
blob: BlobId::hash_of(glb),
|
|
byte_len: glb.len() as u64,
|
|
dims: None,
|
|
},
|
|
}],
|
|
thumbnail: Some(ImportThumbnail {
|
|
path: "previews/watchtower.png".into(),
|
|
meta: ThumbnailMeta {
|
|
blob: BlobId::hash_of(preview),
|
|
media: ThumbnailMedia::Png,
|
|
width: 512,
|
|
height: 512,
|
|
byte_len: preview.len() as u64,
|
|
views: Vec::new(),
|
|
},
|
|
}),
|
|
metrics: Metrics {
|
|
total_bytes: (glb.len() + preview.len()) as u64,
|
|
triangles: 12,
|
|
vertices: 8,
|
|
joints: 0,
|
|
clips: 0,
|
|
max_texture_dim: 0,
|
|
media_millis: 0,
|
|
},
|
|
coordinate_system: CoordinateSystem {
|
|
units_per_meter: 1.0,
|
|
up: Axis::YPos,
|
|
forward: Axis::ZNeg,
|
|
pivot: Pivot::Origin,
|
|
},
|
|
bounds: Bounds {
|
|
min: Vec3::new(-1.0, -1.0, -1.0),
|
|
max: Vec3::new(1.0, 1.0, 1.0),
|
|
},
|
|
anchors: vec![],
|
|
capabilities: Capabilities::default(),
|
|
spawn_recipe: None,
|
|
}],
|
|
rights: fixture_rights(),
|
|
};
|
|
manifest.canonicalize();
|
|
manifest
|
|
}
|
|
|
|
fn fixture_tool() -> ToolClosure {
|
|
ToolClosure {
|
|
processor: "mp_derive".into(),
|
|
version: "1.0".into(),
|
|
build: "deadbeef".into(),
|
|
deterministic: true,
|
|
}
|
|
}
|
|
|
|
fn fixture_thumb_variant(base: AssetRevisionRef) -> DerivedVariantManifest {
|
|
DerivedVariantManifest {
|
|
base,
|
|
kind: RecipeKind::MeshThumbnail,
|
|
recipe: ProcessingRecipe {
|
|
settings: RecipeSettings::MeshThumbnail {
|
|
width: 512,
|
|
height: 512,
|
|
media: ThumbnailMedia::Png,
|
|
},
|
|
tool: fixture_tool(),
|
|
output_schema: OUTPUT_SCHEMA_V1,
|
|
}
|
|
.digest()
|
|
.unwrap(),
|
|
inputs: vec![DerivedInput {
|
|
role: FileRole::RenderGlb,
|
|
blob: BlobId::hash_of(b"watchtower-glb"),
|
|
}],
|
|
outputs: vec![],
|
|
thumbnail: Some(ThumbnailMeta {
|
|
blob: BlobId::hash_of(b"derived-thumb"),
|
|
media: ThumbnailMedia::Png,
|
|
width: 512,
|
|
height: 512,
|
|
byte_len: 450,
|
|
views: Vec::new(),
|
|
}),
|
|
metrics: Metrics {
|
|
total_bytes: 450,
|
|
..Default::default()
|
|
},
|
|
rights: fixture_rights(),
|
|
}
|
|
}
|
|
|
|
fn fixture_profile() -> ClientProfile {
|
|
ClientProfile {
|
|
policy_version: RESOLUTION_POLICY_V1,
|
|
tier: DeviceTier::High,
|
|
max_texture_dim: 2048,
|
|
max_triangles: 1_000_000,
|
|
max_variant_bytes: 64 * 1024 * 1024,
|
|
accept_png: true,
|
|
accept_jpeg: true,
|
|
accept_glb: true,
|
|
accept_bin: true,
|
|
}
|
|
}
|
|
|
|
fn import_config(name: &str, token: &str) -> ClientConfig {
|
|
let mut cfg = config(name);
|
|
cfg.token = Some(token.to_string());
|
|
cfg
|
|
}
|
|
|
|
fn connect_import(name: &str, plane: &ImportPlane, token: &str) -> AssetClient {
|
|
AssetClient::connect(import_config(name, token), plane.endpoints(), None).unwrap()
|
|
}
|
|
|
|
#[test]
|
|
fn import_source_register_list_and_idempotent_retry() {
|
|
let plane = ImportPlane::start();
|
|
let client = connect_import("src_reg", &plane, &plane.write_token);
|
|
let bytes = fixture_collection().to_canonical_bytes().unwrap();
|
|
let expected = SourceCollectionId::hash_of(&bytes);
|
|
let first = client.register_source_collection(&bytes).unwrap();
|
|
assert_eq!(first.source_id, "kenney");
|
|
assert_eq!(first.digest, expected);
|
|
let retry = client.register_source_collection(&bytes).unwrap();
|
|
assert_eq!(retry, first);
|
|
let listed = client.list_source_collections().unwrap();
|
|
assert_eq!(listed.len(), 1);
|
|
assert_eq!(listed[0].source_id, "kenney");
|
|
assert_eq!(listed[0].digest, expected);
|
|
assert_eq!(listed[0].license, "CC0-1.0");
|
|
}
|
|
|
|
#[test]
|
|
fn import_run_status_and_idempotent_retry() {
|
|
let plane = ImportPlane::start();
|
|
let client = connect_import("imp_run", &plane, &plane.write_token);
|
|
let collection = fixture_collection().to_canonical_bytes().unwrap();
|
|
client.register_source_collection(&collection).unwrap();
|
|
let manifest = fixture_import();
|
|
let bytes = manifest.to_canonical_bytes().unwrap();
|
|
let expected = ImportRevisionId::hash_of(&bytes);
|
|
let first = client.run_import(&bytes).unwrap();
|
|
assert!(first.created);
|
|
assert_eq!(first.import_revision, expected);
|
|
assert_eq!(first.entries.len(), 1);
|
|
assert_eq!(first.entries[0].key.as_str(), "models/watchtower");
|
|
assert_eq!(first.entries[0].asset_id, manifest.asset_id_for(&manifest.assets[0].key));
|
|
let retry = client.run_import(&bytes).unwrap();
|
|
assert!(!retry.created);
|
|
assert_eq!(retry.import_revision, first.import_revision);
|
|
assert_eq!(retry.entries, first.entries);
|
|
let status = client.import_status(&expected).unwrap();
|
|
assert_eq!(status.import_revision, expected);
|
|
assert_eq!(status.source_id, "kenney");
|
|
assert_eq!(status.pack_name, "space-kit");
|
|
assert_eq!(status.entries[0].asset_id, first.entries[0].asset_id);
|
|
}
|
|
|
|
#[test]
|
|
fn derived_variant_and_variant_set_identities() {
|
|
let plane = ImportPlane::start();
|
|
let mut client = connect_import("dvar", &plane, &plane.write_token);
|
|
let base = AssetRevisionRef {
|
|
asset_id: AssetId::from_bytes([0x10; 16]),
|
|
revision: AssetRevisionId::from_bytes([0x20; 32]),
|
|
};
|
|
let variant = fixture_thumb_variant(base);
|
|
let variant_bytes = variant.to_canonical_bytes().unwrap();
|
|
let variant_id = plane.seed_variant(variant_bytes);
|
|
assert_eq!(variant_id, variant.id().unwrap());
|
|
let fetched = client.fetch_derived_variant(&variant_id).unwrap();
|
|
assert_eq!(fetched, variant);
|
|
let set_id = client.freeze_variant_set(&base, &[variant_id]).unwrap();
|
|
let retry = client.freeze_variant_set(&base, &[variant_id]).unwrap();
|
|
assert_eq!(retry, set_id);
|
|
let set = client.fetch_variant_set(&set_id).unwrap();
|
|
assert_eq!(set.base, base);
|
|
assert_eq!(set.variants, vec![variant_id]);
|
|
let map = client.resolve_variant_set(&set_id, &fixture_profile()).unwrap();
|
|
assert_eq!(map.set, set_id);
|
|
assert_eq!(map.profile, fixture_profile().digest().unwrap());
|
|
assert!(!map.entries.is_empty());
|
|
}
|
|
|
|
#[test]
|
|
fn import_derived_authorization() {
|
|
let plane = ImportPlane::start();
|
|
let collection = fixture_collection().to_canonical_bytes().unwrap();
|
|
let import_bytes = fixture_import().to_canonical_bytes().unwrap();
|
|
let writer = connect_import("auth_w", &plane, &plane.write_token);
|
|
writer.register_source_collection(&collection).unwrap();
|
|
writer.run_import(&import_bytes).unwrap();
|
|
|
|
let reader = connect_import("auth_r", &plane, &plane.read_token);
|
|
assert!(reader.list_source_collections().is_ok());
|
|
let irev = ImportRevisionId::hash_of(&import_bytes);
|
|
assert!(reader.import_status(&irev).is_ok());
|
|
match reader.register_source_collection(&collection) {
|
|
Err(ClientError::Denied) => {}
|
|
other => panic!("register must deny reader, got {other:?}"),
|
|
}
|
|
match reader.run_import(&import_bytes) {
|
|
Err(ClientError::Denied) => {}
|
|
other => panic!("import run must deny reader, got {other:?}"),
|
|
}
|
|
let base = AssetRevisionRef {
|
|
asset_id: AssetId::from_bytes([0x10; 16]),
|
|
revision: AssetRevisionId::from_bytes([0x20; 32]),
|
|
};
|
|
match reader.freeze_variant_set(&base, &[DerivedVariantId::from_bytes([1; 32])]) {
|
|
Err(ClientError::Denied) => {}
|
|
other => panic!("freeze must deny reader, got {other:?}"),
|
|
}
|
|
|
|
let anon = makepad_asset_client::Api::new(
|
|
plane.endpoints(),
|
|
fast_limits(),
|
|
None,
|
|
)
|
|
.unwrap();
|
|
match anon.list_source_collections() {
|
|
Err(ClientError::Unauthenticated) => {}
|
|
other => panic!("anonymous list must 401, got {other:?}"),
|
|
}
|
|
match anon.run_import(&import_bytes) {
|
|
Err(ClientError::Unauthenticated) => {}
|
|
other => panic!("anonymous import must 401, got {other:?}"),
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn import_derived_tampered_identities_fail_closed() {
|
|
let plane = ImportPlane::start();
|
|
let client = connect_import("tamper", &plane, &plane.write_token);
|
|
let collection = fixture_collection().to_canonical_bytes().unwrap();
|
|
let import_bytes = fixture_import().to_canonical_bytes().unwrap();
|
|
let base = AssetRevisionRef {
|
|
asset_id: AssetId::from_bytes([0x10; 16]),
|
|
revision: AssetRevisionId::from_bytes([0x20; 32]),
|
|
};
|
|
let variant = fixture_thumb_variant(base);
|
|
let variant_id = plane.seed_variant(variant.to_canonical_bytes().unwrap());
|
|
|
|
plane.set_tamper(ImportTamper::SourceDigest);
|
|
match client.register_source_collection(&collection) {
|
|
Err(ClientError::DigestMismatch { what, .. }) => {
|
|
assert_eq!(what, "source collection digest")
|
|
}
|
|
other => panic!("tampered source digest must refuse, got {other:?}"),
|
|
}
|
|
plane.set_tamper(ImportTamper::None);
|
|
client.register_source_collection(&collection).unwrap();
|
|
|
|
plane.set_tamper(ImportTamper::ImportRevision);
|
|
match client.run_import(&import_bytes) {
|
|
Err(ClientError::DigestMismatch { what, .. }) => assert_eq!(what, "import revision"),
|
|
other => panic!("tampered import revision must refuse, got {other:?}"),
|
|
}
|
|
plane.set_tamper(ImportTamper::ImportEntryRevision);
|
|
match client.run_import(&import_bytes) {
|
|
Err(ClientError::DigestMismatch { what, .. }) => {
|
|
assert_eq!(what, "import entry revision")
|
|
}
|
|
other => panic!("tampered entry revision must refuse, got {other:?}"),
|
|
}
|
|
plane.set_tamper(ImportTamper::None);
|
|
let report = client.run_import(&import_bytes).unwrap();
|
|
|
|
plane.set_tamper(ImportTamper::ImportGetId);
|
|
match client.import_status(&report.import_revision) {
|
|
Err(ClientError::Protocol { what }) => assert_eq!(what, "import status id mismatch"),
|
|
other => panic!("tampered import get id must refuse, got {other:?}"),
|
|
}
|
|
plane.set_tamper(ImportTamper::None);
|
|
|
|
let mut fetch_client = connect_import("tamper_fetch", &plane, &plane.write_token);
|
|
plane.set_tamper(ImportTamper::VariantBytes);
|
|
match fetch_client.fetch_derived_variant(&variant_id) {
|
|
Err(ClientError::DigestMismatch { what, .. }) => {
|
|
assert_eq!(what, "derived variant bytes")
|
|
}
|
|
other => panic!("tampered variant bytes must refuse, got {other:?}"),
|
|
}
|
|
plane.set_tamper(ImportTamper::VariantEtag);
|
|
match fetch_client.fetch_derived_variant(&variant_id) {
|
|
Err(ClientError::Protocol { what }) => assert_eq!(what, "canonical etag mismatch"),
|
|
other => panic!("tampered variant etag must refuse, got {other:?}"),
|
|
}
|
|
plane.set_tamper(ImportTamper::VariantSetId);
|
|
match client.freeze_variant_set(&base, &[variant_id]) {
|
|
Err(ClientError::DigestMismatch { what, .. }) => {
|
|
assert_eq!(what, "variant set identity")
|
|
}
|
|
other => panic!("tampered variant set id must refuse, got {other:?}"),
|
|
}
|
|
plane.set_tamper(ImportTamper::None);
|
|
let set_id = client.freeze_variant_set(&base, &[variant_id]).unwrap();
|
|
plane.set_tamper(ImportTamper::ResolveDigest);
|
|
match client.resolve_variant_set(&set_id, &fixture_profile()) {
|
|
Err(ClientError::DigestMismatch { what, .. }) => {
|
|
assert_eq!(what, "resolution map digest")
|
|
}
|
|
other => panic!("tampered resolve digest must refuse, got {other:?}"),
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn import_derived_runtime_request_output_mapping() {
|
|
let plane = ImportPlane::start();
|
|
let client = connect_import("rt_map", &plane, &plane.write_token);
|
|
let collection = fixture_collection().to_canonical_bytes().unwrap();
|
|
let import_bytes = fixture_import().to_canonical_bytes().unwrap();
|
|
let base = AssetRevisionRef {
|
|
asset_id: AssetId::from_bytes([0x10; 16]),
|
|
revision: AssetRevisionId::from_bytes([0x20; 32]),
|
|
};
|
|
let variant = fixture_thumb_variant(base);
|
|
let variant_id = plane.seed_variant(variant.to_canonical_bytes().unwrap());
|
|
|
|
let mut runtime = ClientRuntime::start(client).unwrap();
|
|
let id_reg = runtime
|
|
.submit(ClientRequest::RegisterSourceCollection {
|
|
bytes: collection.clone(),
|
|
})
|
|
.unwrap();
|
|
match wait_runtime(&mut runtime, id_reg).unwrap() {
|
|
ClientOutput::SourceCollectionRegistered(SourceCollectionRegistered { source_id, .. }) => {
|
|
assert_eq!(source_id, "kenney")
|
|
}
|
|
other => panic!("unexpected register output {other:?}"),
|
|
}
|
|
let id_list = runtime.submit(ClientRequest::ListSourceCollections).unwrap();
|
|
match wait_runtime(&mut runtime, id_list).unwrap() {
|
|
ClientOutput::SourceCollections(rows) => assert_eq!(rows.len(), 1),
|
|
other => panic!("unexpected list output {other:?}"),
|
|
}
|
|
let id_run = runtime
|
|
.submit(ClientRequest::RunImport {
|
|
bytes: import_bytes.clone(),
|
|
})
|
|
.unwrap();
|
|
let report = match wait_runtime(&mut runtime, id_run).unwrap() {
|
|
ClientOutput::ImportReport(r) => r,
|
|
other => panic!("unexpected import output {other:?}"),
|
|
};
|
|
let id_status = runtime
|
|
.submit(ClientRequest::FetchImport {
|
|
revision: report.import_revision,
|
|
})
|
|
.unwrap();
|
|
match wait_runtime(&mut runtime, id_status).unwrap() {
|
|
ClientOutput::ImportStatus(s) => assert_eq!(s.import_revision, report.import_revision),
|
|
other => panic!("unexpected status output {other:?}"),
|
|
}
|
|
let id_var = runtime
|
|
.submit(ClientRequest::FetchDerivedVariant { id: variant_id })
|
|
.unwrap();
|
|
match wait_runtime(&mut runtime, id_var).unwrap() {
|
|
ClientOutput::DerivedVariant(v) => assert_eq!(v.id().unwrap(), variant_id),
|
|
other => panic!("unexpected derived output {other:?}"),
|
|
}
|
|
let id_freeze = runtime
|
|
.submit(ClientRequest::FreezeVariantSet {
|
|
base,
|
|
variants: vec![variant_id],
|
|
})
|
|
.unwrap();
|
|
let set_id = match wait_runtime(&mut runtime, id_freeze).unwrap() {
|
|
ClientOutput::VariantSetFrozen(id) => id,
|
|
other => panic!("unexpected freeze output {other:?}"),
|
|
};
|
|
let id_set = runtime
|
|
.submit(ClientRequest::FetchVariantSet { id: set_id })
|
|
.unwrap();
|
|
match wait_runtime(&mut runtime, id_set).unwrap() {
|
|
ClientOutput::VariantSet(set) => assert_eq!(set.id().unwrap(), set_id),
|
|
other => panic!("unexpected set output {other:?}"),
|
|
}
|
|
let id_res = runtime
|
|
.submit(ClientRequest::ResolveVariantSet {
|
|
set: set_id,
|
|
profile: fixture_profile(),
|
|
})
|
|
.unwrap();
|
|
match wait_runtime(&mut runtime, id_res).unwrap() {
|
|
ClientOutput::ResolvedVariants(map) => assert_eq!(map.set, set_id),
|
|
other => panic!("unexpected resolve output {other:?}"),
|
|
}
|
|
runtime.shutdown();
|
|
}
|
|
|
|
fn seed_n_sources(plane: &ImportPlane, n: usize) {
|
|
for i in 0..n {
|
|
plane.seed_source(SourceCollection {
|
|
id: format!("s{i:03}"),
|
|
title: format!("Source {i:03}"),
|
|
origin: SourceOrigin::Upload,
|
|
terms: fixture_rights(),
|
|
});
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn source_collections_page_cursor_order_and_limit() {
|
|
let plane = ImportPlane::start();
|
|
seed_n_sources(&plane, 5);
|
|
let client = connect_import("src_page", &plane, &plane.write_token);
|
|
|
|
match client.source_collections_page(None, 0) {
|
|
Err(ClientError::InvalidInput { what }) => assert_eq!(what, "source page limit"),
|
|
other => panic!("limit 0 must be local, got {other:?}"),
|
|
}
|
|
match client.source_collections_page(None, 501) {
|
|
Err(ClientError::InvalidInput { what }) => assert_eq!(what, "source page limit"),
|
|
other => panic!("limit 501 must be local, got {other:?}"),
|
|
}
|
|
|
|
let first = client.source_collections_page(None, 2).unwrap();
|
|
assert_eq!(
|
|
first.sources.iter().map(|r| r.source_id.as_str()).collect::<Vec<_>>(),
|
|
["s000", "s001"]
|
|
);
|
|
assert_eq!(first.next.as_ref().map(|c| c.server_id()), Some(client.server_id()).as_ref());
|
|
let second = client
|
|
.source_collections_page(first.next.as_ref(), 2)
|
|
.unwrap();
|
|
assert_eq!(
|
|
second.sources.iter().map(|r| r.source_id.as_str()).collect::<Vec<_>>(),
|
|
["s002", "s003"]
|
|
);
|
|
let last = client
|
|
.source_collections_page(second.next.as_ref(), 2)
|
|
.unwrap();
|
|
assert_eq!(
|
|
last.sources.iter().map(|r| r.source_id.as_str()).collect::<Vec<_>>(),
|
|
["s004"]
|
|
);
|
|
assert!(last.next.is_none());
|
|
|
|
let all = client.list_source_collections().unwrap();
|
|
assert_eq!(all.len(), 5);
|
|
let mut sorted = all.clone();
|
|
sorted.sort_by(|a, b| a.source_id.cmp(&b.source_id));
|
|
assert_eq!(all, sorted);
|
|
}
|
|
|
|
#[test]
|
|
fn source_collections_page_tamper_and_legacy_aggregate() {
|
|
let plane = ImportPlane::start();
|
|
seed_n_sources(&plane, 5);
|
|
let client = connect_import("src_tamper", &plane, &plane.write_token);
|
|
|
|
plane.set_tamper(ImportTamper::SourcePageOrder);
|
|
match client.source_collections_page(None, 3) {
|
|
Err(ClientError::Protocol { what }) => assert_eq!(what, "source collection order"),
|
|
other => panic!("reversed page must refuse, got {other:?}"),
|
|
}
|
|
plane.set_tamper(ImportTamper::SourcePageCursor);
|
|
match client.source_collections_page(None, 2) {
|
|
Err(ClientError::Protocol { what }) => {
|
|
assert!(what == "source page cursor" || what == "source page cursor mismatch")
|
|
}
|
|
other => panic!("tampered cursor must refuse, got {other:?}"),
|
|
}
|
|
plane.set_tamper(ImportTamper::None);
|
|
|
|
seed_n_sources(&plane, 513);
|
|
match client.list_source_collections() {
|
|
Err(ClientError::OverBudget { what, limit, found }) => {
|
|
assert_eq!(what, "source collections");
|
|
assert_eq!(limit, 512);
|
|
assert!(found > 512);
|
|
}
|
|
other => panic!("513 sources must fail closed, not partial, got {other:?}"),
|
|
}
|
|
let exact = ImportPlane::start();
|
|
seed_n_sources(&exact, 512);
|
|
let client512 = connect_import("src_512", &exact, &exact.write_token);
|
|
assert_eq!(client512.list_source_collections().unwrap().len(), 512);
|
|
}
|
|
|
|
#[test]
|
|
fn source_collections_cursor_refuses_another_server() {
|
|
let a = ImportPlane::start_with_id([0x11; 16]);
|
|
let b = ImportPlane::start_with_id([0x22; 16]);
|
|
seed_n_sources(&a, 3);
|
|
seed_n_sources(&b, 3);
|
|
let ca = connect_import("src_srv_a", &a, &a.write_token);
|
|
let cb = connect_import("src_srv_b", &b, &b.write_token);
|
|
assert_ne!(ca.server_id(), cb.server_id());
|
|
let page = ca.source_collections_page(None, 2).unwrap();
|
|
let cursor = page.next.expect("continuation cursor");
|
|
assert_eq!(cursor.server_id(), &ca.server_id());
|
|
match cb.source_collections_page(Some(&cursor), 2) {
|
|
Err(ClientError::WrongServerCursor) => {}
|
|
other => panic!("foreign source cursor must refuse, got {other:?}"),
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn source_page_over_ceiling_is_refused() {
|
|
let cap = makepad_asset_client::wire::MAX_SOURCE_PAGE_JSON_RESPONSE_BYTES;
|
|
let server = RawServer::start(Arc::new(move |_req, stream| {
|
|
let head = response_head(200, "application/json", cap + 1, &[]);
|
|
write_raw(stream, head.as_bytes());
|
|
}));
|
|
let api = makepad_asset_client::Api::new(
|
|
makepad_asset_client::ApiEndpoints {
|
|
control: server.addr,
|
|
data: server.addr,
|
|
},
|
|
fast_limits(),
|
|
None,
|
|
)
|
|
.unwrap();
|
|
match api.source_collections_page(None, 1) {
|
|
Err(ClientError::OverBudget { what, limit, found }) => {
|
|
assert_eq!(what, "json response body");
|
|
assert_eq!(limit, cap);
|
|
assert_eq!(found, cap + 1);
|
|
}
|
|
other => panic!("over-ceiling source page must refuse, got {other:?}"),
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// side-channels: stems + lyrics attached to a published audio asset
|
|
// ---------------------------------------------------------------------------
|
|
|
|
#[test]
|
|
fn side_channels_attach_reuse_blobs_and_are_idempotent() {
|
|
use makepad_asset_client::side_channels::{SideChannelFile, SideChannelOutcome};
|
|
use makepad_asset_client::{PublishFile, PublishRequest, PublishThumbnail};
|
|
use makepad_asset_data::{AssetKind, MediaType, ThumbnailMedia};
|
|
|
|
let token = format!("mpat_{}", "6d".repeat(32));
|
|
let fx = FixtureServer::start(
|
|
FixtureStore::default(),
|
|
FixtureOptions { auth_token: Some(token.clone()), ..FixtureOptions::default() },
|
|
);
|
|
let mut cfg = config("side_channels");
|
|
cfg.token = Some(token);
|
|
let mut client = AssetClient::connect(cfg, fx.endpoints(), None).expect("connect");
|
|
|
|
// A published audio track.
|
|
let audio = payload(41, 9_000);
|
|
let request = PublishRequest::new(
|
|
"music",
|
|
AssetKind::Audio,
|
|
"Test Track",
|
|
PublishFile {
|
|
bytes: audio.clone(),
|
|
media: MediaType::Mp3,
|
|
role: FileRole::Audio,
|
|
media_millis: 187_000,
|
|
dims: None,
|
|
},
|
|
PublishThumbnail {
|
|
bytes: payload(42, 900),
|
|
media: ThumbnailMedia::Png,
|
|
width: 512,
|
|
height: 512,
|
|
views: Vec::new(),
|
|
},
|
|
);
|
|
let published = client.publish_artifact(&request).expect("publish audio");
|
|
|
|
// Attach the four stems and the lyrics.
|
|
let stems: Vec<Vec<u8>> = (0..4).map(|i| payload(50 + i, 4_000 + i as usize)).collect();
|
|
let lyrics = br#"{"format":"vj-lyrics","version":4,"lines":[]}"#.to_vec();
|
|
let files = |stems: &[Vec<u8>], lyrics: &[u8]| -> Vec<SideChannelFile> {
|
|
let mut out: Vec<SideChannelFile> = FileRole::STEMS
|
|
.iter()
|
|
.zip(stems)
|
|
.map(|(role, bytes)| SideChannelFile {
|
|
role: *role,
|
|
media: MediaType::Ogg,
|
|
bytes: bytes.clone(),
|
|
})
|
|
.collect();
|
|
out.push(SideChannelFile {
|
|
role: FileRole::Lyrics,
|
|
media: MediaType::Json,
|
|
bytes: lyrics.to_vec(),
|
|
});
|
|
out
|
|
};
|
|
let outcome = client
|
|
.publish_side_channel_files(&published.asset_id, files(&stems, &lyrics))
|
|
.expect("attach side channels");
|
|
let revision = match outcome {
|
|
SideChannelOutcome::Published { revision } => revision,
|
|
other => panic!("expected a publish, got {other:?}"),
|
|
};
|
|
assert_ne!(revision, published.revision, "a NEW revision is the head");
|
|
|
|
// The head advanced and the manifest carries every role, with the audio
|
|
// blob REUSED (same id, no second upload of its bytes).
|
|
let detail = client.asset_detail(&published.asset_id).expect("detail");
|
|
assert_eq!(detail.latest_published().unwrap().revision, revision);
|
|
let manifest = client.fetch_asset_manifest(&revision).expect("manifest");
|
|
for role in FileRole::STEMS {
|
|
assert!(manifest.files.iter().any(|f| f.role == role), "{role:?} present");
|
|
}
|
|
assert!(manifest.files.iter().any(|f| f.role == FileRole::Lyrics));
|
|
let audio_file =
|
|
manifest.files.iter().find(|f| f.role == FileRole::Audio).expect("audio kept");
|
|
assert_eq!(audio_file.blob, published.artifact_blob, "audio blob reused");
|
|
// Total bytes still exactly the sum of the parts (validate enforces it,
|
|
// but assert the value moved).
|
|
let sum: u64 = manifest.files.iter().map(|f| f.byte_len).sum::<u64>()
|
|
+ manifest.thumbnail.as_ref().map(|t| t.byte_len).unwrap_or(0);
|
|
assert_eq!(manifest.metrics.total_bytes, sum);
|
|
|
|
// A stem blob round-trips through the verified cache path.
|
|
let stem_file =
|
|
manifest.files.iter().find(|f| f.role == FileRole::StemDrums).expect("drums");
|
|
let bytes =
|
|
client.fetch_blob_bytes(&stem_file.blob, Some(stem_file.byte_len)).expect("stem blob");
|
|
assert_eq!(bytes, stems[0]);
|
|
|
|
// Idempotence: a second identical attach is a no-op reporting the head.
|
|
let again = client
|
|
.publish_side_channel_files(&published.asset_id, files(&stems, &lyrics))
|
|
.expect("re-attach");
|
|
assert_eq!(again, SideChannelOutcome::AlreadyPresent { revision });
|
|
|
|
// A PARTIAL stem set refuses locally (contract: all four or none) on a
|
|
// FRESH asset — on the analysed one above it would be absorbed by role
|
|
// idempotence instead, which is also what a re-bake relies on.
|
|
let mut fresh = request.clone();
|
|
fresh.title = "Second Track".into();
|
|
fresh.artifact.bytes[0] ^= 0x55;
|
|
let second = client.publish_artifact(&fresh).expect("publish second audio");
|
|
let posts_before = fx.log.requests.lock().unwrap().len();
|
|
let partial: Vec<SideChannelFile> = FileRole::STEMS[..3]
|
|
.iter()
|
|
.zip(&stems)
|
|
.map(|(role, bytes)| SideChannelFile {
|
|
role: *role,
|
|
media: MediaType::Ogg,
|
|
bytes: bytes.iter().map(|b| b ^ 1).collect(),
|
|
})
|
|
.collect();
|
|
let err = client.publish_side_channel_files(&second.asset_id, partial);
|
|
assert!(err.is_err(), "partial stems must refuse: {err:?}");
|
|
let log = fx.log.requests.lock().unwrap();
|
|
assert!(
|
|
!log[posts_before..].iter().any(|r| r.method == "POST" && r.target.contains("/revisions")),
|
|
"no staging happened for the refused set"
|
|
);
|
|
}
|