2d23dba736
The asset store now uses libs/sqlite_query as its ONLY engine — not a feature
flag, not a fallback. That closes the Windows gap (the embedded store starts
there now, and a SHARED->EXCLUSIVE upgrade is handled rather than assumed
free) and takes the C dependency out of the build everywhere else.
Around it:
- store: a garbage collector, catalogued content that is referenced in place
instead of copied, the `vjeffect` kind, and host/chat routes that keep up
with the chat wire below.
- importer: the unified map contract reaches quake2, quake3, doom and duke —
world placement, nav, welding, prelit maps and glTF node handling shared
rather than reimplemented per game. Music import, billboards and stateful
props move to the data crate so readers stop linking the importer.
- ai: the serving side of multi-lane chat — per-lane conversations, honest
progress and acceptance reporting, penalties and a watchdog, context as a
per-box number that compacts instead of erupting, a realtime session mode,
and inpaint/flux2 backends. `chat_bench` measures the rate the way the
client meter computes it.
- client / chat / chat_ui: a publication can NAME a file instead of carrying
it; the wire says whether a turn is warm and whether it is thinking, so a
client stops guessing; transcript and feed widgets render history the way
the model wrote it. `SessionConfig::catalog_runtime` lets a host size the
catalog runtime's lanes itself — a browsing UI puts every listing, every
per-tile resolve and every thumbnail blob through that one runtime and
wants a wider fast lane than the shared default, while media lanes keep
it (a few big transfers, not a thousand small ones).
- widgets: the shared asset widgets — one video view (knobbed seek,
transport, bracket trim, rail playback) used everywhere, plus thumb,
preview, scene view, walk-world and the lyric reader.
2499 lines
98 KiB
Rust
2499 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.
|
|
|
|
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.path).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.path.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.path, resolved.path);
|
|
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.path, resolved.path);
|
|
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 { path, .. } => {
|
|
done_path = Some(path);
|
|
}
|
|
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, path } => {
|
|
assert_eq!(got, blob);
|
|
assert_eq!(std::fs::read(path).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"
|
|
);
|
|
}
|