makepad/libs/asset/store/examples/store_scan.rs
Admin d24cbee60d asset platform: store pipelines with honest banded progress, vision annotation through the job queue, fleet scheduling (roles, spread, peer provisioning), the music contract, synonym search, the runs card
Squashed from work; the fine-grained history is under tag archive/work-2026-08-29:
- mp* wave: mpwm window manager + the mp app family, WM API, theme bridge, PDF engine fix
- work: land the sources the last commits reference
- kenney: catalogue all 50 free 3D kits; Modal dismissed() never fired
- asset store: central vision-annotation queue; Kenney donate prompt
- asset-ai: vision domain — image + prompt -> text on every fleet node
- annotation runs the fleet's vision services through the normal job queue
- vj: responsive DJ mixer + Windows drag-and-drop, cherry-picked from PR #1199 (vjroger)
- store search: WordNet synonym expansion, query-side; per-term seeks
- video scheduling: a cold model pin never downloads past a warm one, and a dedicated box only serves its role
- an expansion can never lose a run — and `expand: true` finally means something
- fleet panel: a slow box is not a missing box
- h3: first+last keyframe conditioning (the weights were always FL2VA)
- registry: the three H3 FL2VA tiers name their real conditioning
- fleet scheduling: spread before stacking, and a stuck job moves
- a download never steals a job from a box that has the weights
- the faster GPU takes the tie: 6000 > 5090 > 4090
- an evicted flux model gives the card its VRAM back
- a job now records what each of its stages was handed
- open a stage in RUNS and read what it sent
- every expanded song was an instrumental: the lyrics had nowhere to go
- the writer never overwrites words the person wrote
- store: a dependent job's body is spliced from its deps' results at claim
- a fleet box's job row says what that job was asked for
- the store runs a whole pipeline, and one record says how far it got
- a client can declare a whole run, watch it, and stop it
- 100% now means published, and the expander is a job you can queue
- one card says what a spawned task is doing, everywhere
- a run that ends says so, instead of being noticed later
- when every box holding the model is busy, buy another copy
2026-08-29 09:26:25 +02:00

191 lines
6.9 KiB
Rust

//! Open a server root read-mostly and prove it is healthy: schema version,
//! CAS hash-path fan-out, browse paging, alias resolution, manifest decode
//! and verified blob reads.
//!
//! Opening a root runs the schema migration, so this is also how a v7 root
//! (one-level CAS paths, no scale indices) is migrated and checked. Point it
//! at a COPY of a live root, never at one a server is using.
//!
//! cargo run --release -p makepad-asset-store --example store_scan -- <root>
use makepad_asset_data::{AssetAlias, AssetManifest};
use makepad_asset_store::{
AssetServerCore, Budgets, SearchFilters, SearchQuery, SearchViewer, ViewerScope,
};
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use std::time::Instant;
fn main() {
let root = match std::env::args().nth(1) {
Some(r) => PathBuf::from(r),
None => {
eprintln!("usage: store_scan <server-root>");
std::process::exit(2);
}
};
let t0 = Instant::now();
let core = AssetServerCore::open(&root, Budgets::default_v1()).expect("open root");
println!("open + migrate: {:?}", t0.elapsed());
println!(
"schema user_version: {}",
makepad_asset_store::SERVER_SCHEMA_VERSION
);
cas_layout(&root.join("cas/objects"));
// Browse the catalog through the same keyset paging the routes use.
let viewer = SearchViewer { principal: None, scope: ViewerScope::All };
let query =
SearchQuery {
text: "",
filters: SearchFilters::default(),
expand: false,
page_size: 100,
facets: 0,
};
let t = Instant::now();
let page = core.search().search(&query, &viewer, None).expect("browse page 1");
println!(
"browse page 1: {} hits of {} total in {:?}",
page.hits.len(),
page.total,
t.elapsed()
);
let t = Instant::now();
let mut walked = page.hits.len();
let mut cursor = page.cursor.clone();
let mut pages = 1;
while let Some(c) = cursor {
let next = core.search().search(&query, &viewer, Some(&c)).expect("browse page n");
walked += next.hits.len();
pages += 1;
cursor = next.cursor;
if pages >= 40 {
break;
}
}
println!("walked {walked} hits over {pages} keyset pages in {:?}", t.elapsed());
// Resolve aliases and read every blob of the first few assets, verified.
let mut resolved = 0;
let mut blobs = 0;
let mut bytes = 0u64;
let t = Instant::now();
for hit in page.hits.iter().take(25) {
let Some(alias) = hit.alias.as_deref() else { continue };
let alias = AssetAlias::new(alias).expect("stored alias parses");
let target = core
.catalog()
.resolve_asset_alias(&alias)
.expect("resolve")
.expect("alias head exists");
resolved += 1;
let manifest_bytes = core
.catalog()
.asset_revision_manifest(&target.revision)
.expect("manifest read")
.expect("head revision stored");
let manifest = AssetManifest::from_canonical_bytes(&manifest_bytes).expect("manifest");
for f in &manifest.files {
let data = core.read_blob(&f.blob).expect("verified blob read");
assert_eq!(data.len() as u64, f.byte_len, "declared byte_len");
blobs += 1;
bytes += data.len() as u64;
}
if let Some(t) = &manifest.thumbnail {
let data = core.read_blob(&t.blob).expect("verified thumbnail read");
blobs += 1;
bytes += data.len() as u64;
}
}
println!(
"resolved {resolved} aliases, read+verified {blobs} blobs ({bytes} bytes) in {:?}",
t.elapsed()
);
// Content-dedup census over every alias head: how many blob references
// the live manifests make, how many distinct blobs that is, and what the
// references would weigh if each one were its own file.
let mut refs = 0u64;
let mut declared = 0u64;
let mut distinct: BTreeMap<[u8; 32], u64> = BTreeMap::new();
let mut walked_aliases = 0u64;
let mut cursor = None;
loop {
let page = core.search().search(&query, &viewer, cursor.as_deref()).expect("browse");
for hit in &page.hits {
let Some(alias) = hit.alias.as_deref() else { continue };
let Ok(alias) = AssetAlias::new(alias) else { continue };
let Some(target) = core.catalog().resolve_asset_alias(&alias).expect("resolve") else {
continue;
};
walked_aliases += 1;
let Some(bytes) = core
.catalog()
.asset_revision_manifest(&target.revision)
.expect("manifest read")
else {
continue;
};
let manifest = AssetManifest::from_canonical_bytes(&bytes).expect("manifest");
let mut note = |blob: &makepad_asset_data::BlobId, len: u64| {
refs += 1;
declared += len;
distinct.insert(*blob.as_bytes(), len);
};
for f in &manifest.files {
note(&f.blob, f.byte_len);
}
if let Some(t) = &manifest.thumbnail {
note(&t.blob, t.byte_len);
}
}
cursor = page.cursor;
if cursor.is_none() {
break;
}
}
let unique_bytes: u64 = distinct.values().sum();
println!(
"dedup census over {walked_aliases} alias heads: {refs} blob references, \
{} distinct blobs, declared {declared} bytes, distinct {unique_bytes} bytes, \
saved {} bytes ({:.1}%)",
distinct.len(),
declared - unique_bytes,
100.0 * (declared - unique_bytes) as f64 / declared.max(1) as f64,
);
println!("OK");
}
/// Directory fan-out of the CAS: depth, entries per level, biggest leaf.
fn cas_layout(objects: &Path) {
let mut per_depth: BTreeMap<usize, u64> = BTreeMap::new();
let mut files = 0u64;
let mut bytes = 0u64;
let mut max_leaf = (0u64, PathBuf::new());
let mut stack = vec![(objects.to_path_buf(), 0usize)];
while let Some((dir, depth)) = stack.pop() {
let mut here = 0u64;
for entry in std::fs::read_dir(&dir).expect("read cas dir") {
let entry = entry.expect("cas entry");
let path = entry.path();
if path.is_dir() {
*per_depth.entry(depth + 1).or_default() += 1;
stack.push((path, depth + 1));
} else {
files += 1;
here += 1;
bytes += entry.metadata().map(|m| m.len()).unwrap_or(0);
}
}
if here > max_leaf.0 {
max_leaf = (here, dir);
}
}
println!(
"cas: {files} objects, {bytes} bytes, dirs per level {per_depth:?}, \
busiest directory {} entries ({:?})",
max_leaf.0, max_leaf.1
);
}