Something went wrong. Try again.
A lexicon-driven AppView for ATProto. happyview.dev
backfill firehose jetstream atproto appview oauth lexicon
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544//! A real `happyview-script` module, read and run by the host: the sections//! the linker actually produced, the capabilities its imports actually need,//! and handlers run through the same execute path an interpreter takes.
use std::collections::BTreeMap;use std::sync::Arc;use std::sync::atomic::{AtomicUsize, Ordering};use std::time::{Duration, Instant};
use serde_json::{Value, json};
use happyview::config::project::ModuleDecl;use happyview::lexicon::{ParsedLexicon, ProcedureAction};use happyview::plugin::capabilities::{PluginCapability, analyze_imports, minimal_set};use happyview::plugin::{ ExecutionError, ScriptErrorKind, ScriptExecuteContext, ScriptExecuteInput, ScriptExecuteLimits, ScriptExecuteOutput, ScriptKind, ScriptLibraryRef, ScriptValueKind, loader,};use happyview::project::script_module::sections::{ModuleHandler, read_sections};use happyview::project::script_module::{ScriptModule, load_modules};use happyview::project::snapshot::ProjectLexicon;use happyview::test_support::{migrated_memory_pool, test_state_with_pool};
const MODULE: &str = "tests/fixtures/script_module/target/wasm32-wasip1/release/script_module.wasm";const SOURCE: &str = "tests/fixtures/script_module/src/lib.rs";
macro_rules! require_fixture { ($path:expr, $build:literal) => { if !std::path::Path::new($path).exists() { eprintln!(concat!("skipping: fixture not built. Run: ", $build)); return; } };}
macro_rules! require_module { () => { require_fixture!( MODULE, "cargo build --manifest-path tests/fixtures/script_module/Cargo.toml --target wasm32-wasip1 --release" ) };}
/// What the module's imports need: the two concurrent-call imports are/// `library:call`, and std's panic machinery links `fd_write`.fn capabilities() -> Vec<PluginCapability> { vec![PluginCapability::LibraryCall, PluginCapability::WasiStdio]}
fn lexicon(id: &str, kind: &str) -> (String, ProjectLexicon) { let main = match kind { "record" => { json!({"type": "record", "key": "tid", "record": {"type": "object", "properties": {}}}) } other => json!({"type": other}), }; let parsed = ParsedLexicon::parse( json!({"lexicon": 1, "id": id, "defs": {"main": main}}), 1, None, ProcedureAction::Upsert, None, ) .unwrap(); ( id.into(), ProjectLexicon { parsed, path: format!("src/lexicons/{id}.json").into(), backfill: false, }, )}
/// The fixture loaded the way a snapshot build loads it.fn module(state: &happyview::AppState) -> Arc<ScriptModule> { let dir = tempfile::tempdir().unwrap(); let build = dir.path().join(".happyview/build"); std::fs::create_dir_all(&build).unwrap(); std::fs::copy(MODULE, build.join("fixture.wasm")).unwrap(); let decls = BTreeMap::from([( "fixture".to_string(), ModuleDecl { language: "rust".into(), path: "tests/fixtures/script_module".into(), capabilities: capabilities(), }, )]); let lexicons = [ lexicon("app.test.echo", "query"), lexicon("app.test.pair", "query"), lexicon("app.test.boom", "query"), lexicon("app.test.spin", "query"), lexicon("app.test.thing", "record"), lexicon("app.test.calibration.shape", "query"), lexicon("app.test.calibration.aggregate", "query"), lexicon("app.test.calibration.render", "query"), ] .into_iter() .collect(); let loaded = load_modules( &state.wasm_runtime, dir.path(), &decls, &lexicons, &BTreeMap::new(), ) .unwrap_or_else(|d| panic!("{d:?}")); loaded.modules.into_values().next().unwrap()}
fn input(trigger: &str, value: Value, instructions: Option<u32>) -> ScriptExecuteInput { ScriptExecuteInput { source: String::new(), kind: ScriptKind::XrpcQuery, input: value, context: ScriptExecuteContext { trigger: trigger.into(), caller_did: Some("did:plc:caller".into()), ..ScriptExecuteContext::default() }, libraries: vec![ScriptLibraryRef { namespace: "http".into(), id: "sdk_http".into(), }], limits: ScriptExecuteLimits { instructions, memory_bytes: 64 * 1024 * 1024, }, removed_globals: Vec::new(), chunk_name: happyview::plugin::DEFAULT_CHUNK_NAME.into(), artifact: None, }}
async fn run( state: &happyview::AppState, module: &ScriptModule, input: &ScriptExecuteInput,) -> Result<ScriptExecuteOutput, ExecutionError> { state .plugin_executor() .execute_compiled_as(module.plugin.clone(), &module.source_path, input, None) .await}
async fn state() -> happyview::AppState { test_state_with_pool(migrated_memory_pool().await)}
#[test]fn the_linker_concatenates_every_handler_line_into_one_readable_section() { require_module!(); let sections = read_sections(&std::fs::read(MODULE).unwrap()).unwrap(); let mut handlers = sections.handlers.clone(); handlers.sort_by(|a, b| a.trigger.cmp(&b.trigger)); let expected: Vec<ModuleHandler> = [ ("record.index:app.test.thing", "index_thing"), ("xrpc.query:app.test.boom", "boom"), ( "xrpc.query:app.test.calibration.aggregate", "calibration_aggregate", ), ( "xrpc.query:app.test.calibration.render", "calibration_render", ), ("xrpc.query:app.test.calibration.shape", "calibration_shape"), ("xrpc.query:app.test.echo", "echo"), ("xrpc.query:app.test.pair", "pair"), ("xrpc.query:app.test.spin", "spin"), ] .into_iter() .map(|(trigger, function)| ModuleHandler { trigger: trigger.into(), function: Some(function.into()), }) .collect(); assert_eq!(handlers, expected); assert_eq!(sections.interface.as_deref(), Some("1")); assert!( !sections.uses_exceptions, "a stable wasm32-wasip1 build must not use exception handling" );}
#[test]fn its_imports_need_exactly_library_call_and_stdio() { require_module!(); let mut needed = minimal_set(&analyze_imports(&std::fs::read(MODULE).unwrap()).unwrap()); needed.sort_by_key(|c| c.as_str()); assert_eq!(needed, capabilities());}
fn manifest() -> happyview::plugin::PluginManifest { serde_json::from_value(json!({ "id": "module:fixture", "name": "fixture", "version": "0.0.0", "api_version": "2", "plugin_type": "interpreter", "language_id": "rust", "capabilities": capabilities(), })) .unwrap()}
#[test]fn dropping_a_capability_it_needs_fails_the_build() { require_module!(); let wasm = std::fs::read(MODULE).unwrap(); // Without this, a manifest refused for some other reason would satisfy // every drop below. loader::validate_capabilities(&manifest(), &wasm).unwrap(); for dropped in capabilities() { let mut manifest = manifest(); manifest.capabilities.retain(|c| *c != dropped); let err = loader::validate_capabilities(&manifest, &wasm).unwrap_err(); assert!( err.to_string().contains(dropped.as_str()), "{dropped:?}: {err}" ); }}
#[tokio::test]async fn a_query_runs_through_the_execute_path() { require_module!(); let state = state().await; let module = module(&state); let output = run( &state, &module, &input( "xrpc.query:app.test.echo", json!({"q": "hi"}), Some(1_000_000), ), ) .await .unwrap(); assert_eq!( output, ScriptExecuteOutput::Returned { value: json!({"q": "hi", "trigger": "xrpc.query:app.test.echo", "caller": "did:plc:caller"}), value_kind: ScriptValueKind::Object, } );}
#[tokio::test]async fn a_record_handler_returning_nothing_reads_as_none() { require_module!(); let state = state().await; let module = module(&state); let mut event = input("record.index:app.test.thing", json!({}), Some(1_000_000)); event.kind = ScriptKind::RecordEvent; let output = run(&state, &module, &event).await.unwrap(); assert_eq!( output, ScriptExecuteOutput::Returned { value: Value::Null, value_kind: ScriptValueKind::None } );}
#[tokio::test]async fn a_panic_is_a_runtime_error_at_the_authors_line() { require_module!(); let source = std::fs::read_to_string(SOURCE).unwrap(); let lines: Vec<u32> = source .lines() .enumerate() .filter(|(_, l)| l.trim() == "panic!(\"the fixture panicked on purpose\")") .map(|(i, _)| i as u32 + 1) .collect(); let [line] = lines[..] else { panic!("expected exactly one panic! line in {SOURCE}, found {lines:?}"); }; let state = state().await; let module = module(&state); let output = run( &state, &module, &input("xrpc.query:app.test.boom", json!({}), Some(1_000_000)), ) .await .unwrap(); let ScriptExecuteOutput::Error { kind, message, line: at, chunk, .. } = output else { panic!("expected an error, got {output:?}"); }; assert_eq!(kind, ScriptErrorKind::Runtime); assert_eq!(chunk.as_deref(), Some(SOURCE)); assert_eq!(at, Some(line)); // The host prints the location beside the message, so the message // carries none of its own. assert_eq!(message, "the fixture panicked on purpose");}
#[tokio::test]async fn spending_the_instruction_budget_is_a_timeout_not_a_crash() { require_module!(); let state = state().await; let module = module(&state); let started = Instant::now(); let err = run( &state, &module, &input("xrpc.query:app.test.spin", json!({}), Some(1_000)), ) .await .unwrap_err(); assert!(matches!(err, ExecutionError::Timeout), "{err}"); // Fuel, not the ten-second wall clock, ended it. assert!( started.elapsed() < Duration::from_secs(5), "{:?}", started.elapsed() );}
#[tokio::test]async fn a_trivial_handler_fits_a_tenth_of_the_default_budget_and_not_one_instruction() { require_module!(); let state = state().await; let module = module(&state); // The echo handler's whole run, decoding its input included, has to fit // in a tenth of the default budget. let output = run( &state, &module, &input( "xrpc.query:app.test.echo", json!({"q": "hi"}), Some(100_000), ), ) .await .unwrap(); assert_eq!( output, ScriptExecuteOutput::Returned { value: json!({"q": "hi", "trigger": "xrpc.query:app.test.echo", "caller": "did:plc:caller"}), value_kind: ScriptValueKind::Object, } ); // Without this, a budget that never reached the guest would pass the // check above. let err = run( &state, &module, &input("xrpc.query:app.test.echo", json!({"q": "hi"}), Some(1)), ) .await .unwrap_err(); assert!(matches!(err, ExecutionError::Timeout), "{err}");}
#[tokio::test]async fn an_unknown_trigger_is_a_missing_handler() { require_module!(); let state = state().await; let module = module(&state); let output = run( &state, &module, &input("xrpc.query:app.test.other", json!({}), Some(1_000_000)), ) .await .unwrap(); let ScriptExecuteOutput::Error { kind: ScriptErrorKind::MissingHandle, message, .. } = &output else { panic!("expected a missing handler, got {output:?}"); }; assert!(message.contains("xrpc.query:app.test.other"), "{message}");}
/// `/slow?ms=N` holds the request for `N` ms and counts how many are in/// flight at once.async fn slow_server() -> (String, Arc<AtomicUsize>) { use axum::extract::{Query, State}; use axum::routing::get;
#[derive(serde::Deserialize)] struct Slow { ms: u64, } #[derive(Default)] struct Seen { now: AtomicUsize, peak: Arc<AtomicUsize>, }
async fn slow(State(seen): State<Arc<Seen>>, Query(q): Query<Slow>) -> &'static str { let now = seen.now.fetch_add(1, Ordering::SeqCst) + 1; seen.peak.fetch_max(now, Ordering::SeqCst); tokio::time::sleep(Duration::from_millis(q.ms)).await; seen.now.fetch_sub(1, Ordering::SeqCst); "slow" }
let seen = Arc::new(Seen::default()); let peak = seen.peak.clone(); let app = axum::Router::new() .route("/slow", get(slow)) .with_state(seen); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let url = format!("http://{}", listener.local_addr().unwrap()); tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); (url, peak)}
#[tokio::test]async fn joined_library_calls_overlap() { require_module!(); require_fixture!( "tests/fixtures/sdk_http/target/wasm32-unknown-unknown/release/sdk_http.wasm", "cargo build --manifest-path tests/fixtures/sdk_http/Cargo.toml --target wasm32-unknown-unknown --release" ); let (base, peak) = slow_server().await; let state = state().await; let http = loader::load_from_file(std::path::Path::new("tests/fixtures/sdk_http")) .await .unwrap(); state.plugin_registry.install(http).await.unwrap(); let module = module(&state);
let output = run( &state, &module, &input( "xrpc.query:app.test.pair", json!({"url": format!("{base}/slow?ms=300")}), None, ), ) .await .unwrap();
let ScriptExecuteOutput::Returned { value, .. } = output else { panic!("{output:?}") }; assert_eq!(value[0]["body"], "slow", "{value}"); assert_eq!(value[1]["body"], "slow", "{value}"); // Each request is held for 300 ms, so two in flight at once is overlap. assert_eq!(peak.load(Ordering::SeqCst), 2);}
#[tokio::test]async fn a_query_reaches_the_module_through_the_xrpc_router() { require_module!(); use tower::ServiceExt;
let dir = tempfile::tempdir().unwrap(); let root = dir.path(); std::fs::write( root.join("happyview.toml"), "[modules.fixture]\nlanguage = \"rust\"\npath = \"modules/fixture\"\ncapabilities = [\"library:call\", \"wasi:stdio\"]\n", ) .unwrap(); std::fs::create_dir_all(root.join("src/lexicons/app/test")).unwrap(); for id in ["echo", "pair", "boom", "spin"] { // `q` is declared so parameter validation passes it through. let main = json!({"type": "query", "parameters": {"type": "params", "properties": { "q": {"type": "string"}, "url": {"type": "string"} }}}); std::fs::write( root.join(format!("src/lexicons/app/test/{id}.json")), json!({"lexicon": 1, "id": format!("app.test.{id}"), "defs": {"main": main}}) .to_string(), ) .unwrap(); } std::fs::create_dir_all(root.join("src/lexicons/app/test/calibration")).unwrap(); for id in ["shape", "aggregate", "render"] { std::fs::write( root.join(format!("src/lexicons/app/test/calibration/{id}.json")), json!({"lexicon": 1, "id": format!("app.test.calibration.{id}"), "defs": {"main": {"type": "query"}}}) .to_string(), ) .unwrap(); } std::fs::write( root.join("src/lexicons/app/test/thing.json"), json!({"lexicon": 1, "id": "app.test.thing", "defs": {"main": {"type": "record", "key": "tid", "record": {"type": "object", "properties": {}}}}}).to_string(), ) .unwrap(); std::fs::create_dir_all(root.join(".happyview/build")).unwrap(); std::fs::copy(MODULE, root.join(".happyview/build/fixture.wasm")).unwrap();
let state = state().await; let report = happyview::project::build::build_snapshot( &state, root, happyview::project::build::BuildOptions { include_tests: false, }, ) .await .unwrap_or_else(|d| panic!("{d:?}")); happyview::project::build::install(&state, report.snapshot).await;
// The router answers only hosts it serves; the bare test state serves none. let now = happyview::db::now_rfc3339(); state .domain_cache .insert(happyview::domain::Domain { id: "test".into(), url: "http://127.0.0.1".into(), is_primary: true, created_at: now.clone(), updated_at: now, }) .await;
let router = happyview::server::router(state.clone()); let response = router .oneshot( axum::http::Request::get("/xrpc/app.test.echo?q=hi") .header("host", "127.0.0.1") .body(axum::body::Body::empty()) .unwrap(), ) .await .unwrap(); assert_eq!(response.status(), 200); let body = http_body_util::BodyExt::collect(response.into_body()) .await .unwrap() .to_bytes(); let body: Value = serde_json::from_slice(&body).unwrap(); assert_eq!(body["q"], "hi"); assert_eq!(body["trigger"], "xrpc.query:app.test.echo");}