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.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211use std::collections::{HashMap, HashSet};use std::hash::{Hash, Hasher};use std::path::{Path, PathBuf};use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};use std::sync::{Arc, Mutex, Weak};use std::time::{Duration, Instant};
use sha2::{Digest, Sha256};use wasmtime::*;
use crate::plugin::{LoadedPlugin, PluginType};
/// Default fuel for plugin execution (≈100ms CPU time). It also has to/// cover instantiation: on a host without a copy-on-write memory image/// (macOS; Linux maps one through memfd) wasmtime charges one fuel per byte/// of initialized data copied in, so a module's data image is bounded by/// this constant there, and instantiation fails outright rather than/// timing out when it is not.pub const DEFAULT_FUEL: u64 = 10_000_000;
/// The fuel a store is given at instantiation and before each call.pub fn fuel_for(plugin_type: PluginType) -> u64 { match plugin_type { PluginType::Library | PluginType::Auth => DEFAULT_FUEL, // Fuel cannot bound an interpreter: wasmtime 49's Cranelift does not // reload the fuel counter in a frame that catches a `throw`, so a // guest that catches an error erases the fuel its body spent, and // every Lua error on the wasi route is such a throw (reproduction: // `.superpowers/spikes/lua-wasm/candidates/eh-probe/fuel_loss.wat`). // Epoch interruption is the boundary that holds, which is why the // engine keeps exceptions enabled: `executor::deadline`'s catching // guest test shows the deadline stopping a `try_table` loop that // fuel does not. Fuel stays enabled engine-wide because it is what // bounds libraries. The whole range also covers the per-byte // instantiation charge a megabytes-sized interpreter image draws on // a host without copy-on-write memory. PluginType::Interpreter => u64::MAX, }}
/// Fuel per unit of a script's instruction budget, for a compiled module.////// The budget is an operator setting written for the Lua interpreter, where a/// unit is one bytecode instruction and everything a script does in a C/// builtin is free: decoding the run's JSON input, encoding its result,/// `table.sort`, `string.format`, `table.concat`. A compiled module does all/// of that in wasm and pays fuel for it, input decoding included, which is/// most of what a typical handler spends. The value therefore sits above the/// top of the measured range, so that every measured workload that fits a Lua/// budget also fits as a module under the same setting; the wall clock/// remains the hard CPU bound for both.////// That does not hold for every handler. A module pays roughly 97 fuel per/// byte to decode its input (the slope of a handler that only decodes, from/// 22,852 to 91,886 bytes of records), which a Lua script does not pay at/// all, so a handler that does little with a large input gets far less from/// one setting as a module: at the setting's floor of 1,000, about 40 KB of/// input exhausts the budget before the handler runs.////// Measured by `tests/fuel_calibration.rs` on 2026-10-13: the smallest budget/// at which each handler returns, as a Lua script under happyview-lua 1.0.0/// (sha256 `9496eb497590e44c5c8a66441156499c63dd6f9c8b9ec01bf59c9b8a5e11382b`)/// and as a release wasm32-wasip1 module in raw fuel, under wasmtime 49.0.2/// and rustc 1.96.1, with module fuel resolved to 100. Three runs of each/// input agreed exactly.////// | handler | records | Lua instructions | module fuel | fuel per instruction |/// |-----------|---------|------------------|-------------|----------------------|/// | shape | 50 | 1,575 | 2,798,400 | 1,777 |/// | aggregate | 50 | 2,321 | 2,461,700 | 1,061 |/// | render | 50 | 852 | 2,748,900 | 3,226 |/// | shape | 200 | 6,429 | 9,899,300 | 1,540 |/// | aggregate | 200 | 7,445 | 9,486,600 | 1,274 |/// | render | 200 | 3,252 | 10,797,700 | 3,320 |/// | echo | - | 37 | 67,700 | 1,830 |////// `shape` filters, sorts and reshapes post envelopes, `aggregate` counts and/// ranks their tags, and `render` formats one line per post. The value is the/// largest ratio, 3,320, plus about a fifth for toolchain drift: a rustc,/// wasmtime or serde upgrade moves these figures by a few percent, and the/// calibration test should fail only on a real shift. That test is ignored/// and needs the plugin, so it runs only by hand: re-run it after upgrading/// rustc, wasmtime, serde or the SDK's input decoding, or happyview-lua. A/// trivial handler needs 17 units under this value, far inside the setting's/// floor of 1,000, and the setting's ceiling of 1,000,000,000 is 4 × 10^12/// fuel, nowhere near overflowing.pub const FUEL_PER_INSTRUCTION: u64 = 4_000;
/// The fuel each call into a compiled module gets. `None` is a job, which/// has no budget and stops through `should_stop`.////// Fuel can bound a compiled module where it cannot bound an interpreter/// (see [`fuel_for`]) because nothing here catches a wasm `throw`: stable/// Rust on wasm32-wasip1 aborts on panic and emits no exception-handling/// instructions, and a module that defines a tag is refused before it runs.pub fn module_fuel(instructions: Option<u32>) -> u64 { instructions.map_or(u64::MAX, |n| { u64::from(n).saturating_mul(FUEL_PER_INSTRUCTION) })}
/// Fuel a guest is granted for each byte the host writes into it — a call's/// input, or an import's answer. Data the host hands over brings the fuel to/// handle it: parsing it, decoding `$bytes`, and encoding it again on the way/// out each cost a few instructions a byte, so a fixed budget would cap every/// payload a library plugin can move at a couple of hundred kilobytes, far/// below what the host itself allows. Computation that brings no data stays/// bounded by [`DEFAULT_FUEL`].////// Measured on `tests/plugin_blobs.rs`'s 4 MiB round trip, where a blob is/// decoded and encoded again in each direction: 64 a byte runs out and 128/// does not. 256 is that with room to spare.pub const FUEL_PER_BYTE_RECEIVED: u64 = 256;
/// Add [`FUEL_PER_BYTE_RECEIVED`] for `len` bytes to a store's remaining/// fuel. A store whose fuel is already unbounded stays so. An interpreter/// store is left alone: an installed interpreter's fuel is unbounded, and a/// compiled module's is the operator's instruction budget, which/// [`FUEL_PER_INSTRUCTION`] already sizes to cover decoding its input.pub fn grant_fuel_for_bytes( store: &mut impl wasmtime::AsContextMut<Data = crate::plugin::host::PluginState>, len: usize,) { let mut store = store.as_context_mut(); if store.data().plugin_type == PluginType::Interpreter { return; } if let Ok(fuel) = store.get_fuel() { let grant = (len as u64).saturating_mul(FUEL_PER_BYTE_RECEIVED); let _ = store.set_fuel(fuel.saturating_add(grant)); }}
/// How often the epoch advances. Against a budget whose floor is one second/// this is a one-percent granularity, for a few hundred atomic increments a/// second.pub const EPOCH_TICK: Duration = Duration::from_millis(10);
/// A deadline no ticker reaches: at one tick per 10 ms it lies billions of/// years out. Half the range rather than all of it because wasmtime adds/// the delta to the current epoch, and that sum must not overflow.pub const UNBOUNDED_TICKS: u64 = u64::MAX / 2;
/// The native stack a guest's frames may take, which wasmtime's default holds/// to 512 KiB. A guest cannot see this stack at all: a C interpreter measures/// the shadow stack in its own linear memory, and each level of its recursion/// costs far more native stack than shadow stack, so the native one runs out/// first and the run ends in a trap rather than the language's own error./// QuickJS-ng's parser is the steepest case measured, at well over twenty/// times; 16 MiB is what lets an interpreter's own limit fire first for every/// shape tried, nested source included, where 8 MiB still trapped on source/// nested five thousand deep. It is reserved per fiber and committed only as/// it is touched, so it costs a run that recurses that deep and nothing else.pub const MAX_WASM_STACK: usize = 16 << 20;
/// Each call runs on a fiber of this size, and what is left above/// [`MAX_WASM_STACK`] is all the stack a host function gets. Wasmtime does not/// bound that: a host function that overruns it aborts the process. 2 MiB is/// more than the 1.5 MiB its defaults left.pub const ASYNC_STACK_SIZE: usize = MAX_WASM_STACK + (2 << 20);
/// WASM runtime for executing pluginspub struct WasmRuntime { engine: Engine, /// Mirrors the engine's epoch tick for tick. Wasmtime exposes neither /// the current epoch nor a store's remaining deadline, and the guest-only /// accounting in [`GuestDeadline`] needs both. epoch: Arc<AtomicU64>, ticker_stop: Arc<AtomicBool>, modules: ModuleCache,}
impl WasmRuntime { pub fn new() -> Result<Self, anyhow::Error> { Self::with_cache_dir(None) }
/// A runtime whose compiled modules are also kept on disk under `dir`. /// Without one, a module compiled in this process outlives only this /// process. pub fn with_cache_dir(dir: Option<PathBuf>) -> Result<Self, anyhow::Error> { let mut runtime = Self::without_ticker()?; runtime.modules.dir = dir.and_then(|dir| match prepare_cache_dir(&dir) { Ok(()) => Some(dir), Err(reason) => { runtime.modules.warned.store(true, Ordering::Relaxed); tracing::warn!( dir = %dir.display(), "{reason}; compiled plugin modules are cached in memory only" ); None } }); let engine = runtime.engine.clone(); let epoch = runtime.epoch.clone(); let stop = runtime.ticker_stop.clone(); std::thread::Builder::new() .name("wasm-epoch".into()) .spawn(move || { // Scheduled against absolute instants so sleep overhead does // not accumulate into a slow clock; a stalled thread catches // up with rapid ticks, which is what wall time means. let mut next = Instant::now() + EPOCH_TICK; while !stop.load(Ordering::Relaxed) { let now = Instant::now(); if next > now { std::thread::sleep(next - now); } next += EPOCH_TICK; tick(&engine, &epoch); } })?; Ok(runtime) }
/// A runtime whose epoch advances only through [`WasmRuntime::tick`], /// for tests that need the clock under their own control. pub fn without_ticker() -> Result<Self, anyhow::Error> { let mut config = Config::new(); config.consume_fuel(true); config.epoch_interruption(true); config.max_wasm_stack(MAX_WASM_STACK); config.async_stack_size(ASYNC_STACK_SIZE);
let engine = Engine::new(&config)?;
Ok(Self { engine, epoch: Arc::new(AtomicU64::new(0)), ticker_stop: Arc::new(AtomicBool::new(false)), modules: ModuleCache::default(), }) }
pub fn engine(&self) -> &Engine { &self.engine }
/// The tick counter every deadline built from this runtime reads. pub fn epoch(&self) -> Arc<AtomicU64> { self.epoch.clone() }
/// Advance the engine's epoch and the mirror by one. pub fn tick(&self) { tick(&self.engine, &self.epoch); }
/// Compile a WASM module pub fn compile(&self, wasm_bytes: &[u8]) -> Result<Module, anyhow::Error> { self.modules.compiles.fetch_add(1, Ordering::Relaxed); Ok(Module::new(&self.engine, wasm_bytes)?) }
/// The compiled module for `plugin`, from memory, from disk, or by /// compiling. The three tiers are checked in that order and a miss at /// any tier fills the ones above it. pub fn module_for(&self, plugin: &Arc<LoadedPlugin>) -> Result<Module, anyhow::Error> { let id = plugin.info.id.as_str(); if let Some(module) = self.modules.get_by_identity(id, plugin) { return Ok(module); } let sha = hex::encode(Sha256::digest(&plugin.wasm_bytes)); if let Some(module) = self.modules.get_by_hash(id, plugin, &sha) { return Ok(module); } let key = self.cache_key(&sha); if let Some(module) = self.modules.load_from_disk(&self.engine, &key) { self.modules.insert(id, plugin, &sha, module.clone()); return Ok(module); } let module = self.compile(&plugin.wasm_bytes)?; self.modules.write_to_disk(&key, &module); self.modules.insert(id, plugin, &sha, module.clone()); Ok(module) }
/// Drop the in-memory module held for `plugin_id`. pub fn evict(&self, plugin_id: &str) { self.modules.evict(plugin_id); }
/// Compiles performed by this runtime, cache hits excluded. pub fn compile_count(&self) -> u64 { self.modules.compiles.load(Ordering::Relaxed) }
/// The file stem a module's serialized form is kept under: the wasm /// bytes' hash and this engine's compatibility fingerprint, which /// covers the wasmtime version and the config. A module compiled by a /// different engine is not loadable by this one, and a stale file must /// read as a miss rather than fail a call. pub fn cache_key(&self, wasm_sha256: &str) -> String { let mut hasher = std::hash::DefaultHasher::new(); self.engine .precompile_compatibility_hash() .hash(&mut hasher); format!("{wasm_sha256}-{:016x}", hasher.finish()) }
/// Delete every serialized module on disk that no installed plugin /// would load: the content-addressed files are the only ones this /// runtime writes, so anything not in `live_wasm_sha256s` under this /// engine's fingerprint is an orphan. pub fn sweep_cache_dir(&self, live_wasm_sha256s: &HashSet<String>) { let Some(dir) = &self.modules.dir else { return; }; let live: HashSet<String> = live_wasm_sha256s .iter() .map(|sha| self.cache_key(sha)) .collect(); let entries = match std::fs::read_dir(dir) { Ok(entries) => entries, Err(e) => { self.modules.warn_once(dir, "read", &e); return; } }; for entry in entries.flatten() { let path = entry.path(); let Some(name) = path.file_name().and_then(|n| n.to_str()) else { continue; }; let Some(stem) = name .strip_suffix(".cwasm") .or_else(|| name.strip_suffix(".cwasm.sha256")) else { continue; }; if live.contains(stem) { continue; } if let Err(e) = std::fs::remove_file(&path) { tracing::warn!(path = %path.display(), error = %e, "could not remove orphaned plugin cache file"); } else { tracing::debug!(path = %path.display(), "removed orphaned plugin cache file"); } } }}
/// Create the cache directory for this process alone, or refuse one that/// is not. The files in it feed `Module::deserialize`, which runs whatever/// it is handed as native code, so a directory another user can write to/// is a directory another user can run code through.fn prepare_cache_dir(dir: &Path) -> Result<(), String> { let mut builder = std::fs::DirBuilder::new(); builder.recursive(true); #[cfg(unix)] { use std::os::unix::fs::DirBuilderExt; builder.mode(0o700); } builder .create(dir) .map_err(|e| format!("could not create the plugin module cache directory: {e}"))?; let metadata = std::fs::metadata(dir) .map_err(|e| format!("could not read the plugin module cache directory: {e}"))?; if !metadata.is_dir() { return Err("the plugin module cache path is not a directory".into()); } #[cfg(unix)] { // SAFETY: `geteuid` reads a process attribute and cannot fail. let euid = unsafe { libc::geteuid() }; directory_is_private(&metadata, euid)?; } Ok(())}
/// A directory is private when this user owns it and nobody else holds a/// permission bit on it. The sidecar hash guards the files against/// corruption; this guards them against an author.#[cfg(unix)]fn directory_is_private(metadata: &std::fs::Metadata, euid: u32) -> Result<(), String> { use crate::fs_private::{NotPrivate, private_to}; private_to(metadata, euid).map_err(|reason| match reason { NotPrivate::OtherOwner { uid, euid } => format!( "the plugin module cache directory is owned by uid {uid} rather than this process's uid {euid}" ), NotPrivate::OpenMode { mode } => format!( "the plugin module cache directory has mode {mode:o}; it must grant nothing to group or other" ), })}
struct CachedModule { /// The registration this module was compiled from. A live match is a /// hit without hashing the wasm bytes; a dead or different one falls /// through to the hash comparison. plugin: Weak<LoadedPlugin>, wasm_sha256: String, module: Module,}
/// Compiled modules by plugin id in memory, and serialized ones by content/// on disk. Everything on disk is best-effort: a directory that cannot be/// created, read or written is one warning and a compile, never a failed/// plugin call.#[derive(Default)]struct ModuleCache { entries: Mutex<HashMap<String, CachedModule>>, dir: Option<PathBuf>, compiles: AtomicU64, warned: AtomicBool,}
impl ModuleCache { fn get_by_identity(&self, id: &str, plugin: &Arc<LoadedPlugin>) -> Option<Module> { let entries = self.entries.lock().unwrap_or_else(|e| e.into_inner()); let entry = entries.get(id)?; let live = entry.plugin.upgrade()?; Arc::ptr_eq(&live, plugin).then(|| entry.module.clone()) }
fn get_by_hash(&self, id: &str, plugin: &Arc<LoadedPlugin>, sha: &str) -> Option<Module> { let mut entries = self.entries.lock().unwrap_or_else(|e| e.into_inner()); let entry = entries.get_mut(id)?; if entry.wasm_sha256 != sha { return None; } entry.plugin = Arc::downgrade(plugin); Some(entry.module.clone()) }
fn insert(&self, id: &str, plugin: &Arc<LoadedPlugin>, sha: &str, module: Module) { self.entries .lock() .unwrap_or_else(|e| e.into_inner()) .insert( id.to_string(), CachedModule { plugin: Arc::downgrade(plugin), wasm_sha256: sha.to_string(), module, }, ); }
fn evict(&self, id: &str) { self.entries .lock() .unwrap_or_else(|e| e.into_inner()) .remove(id); }
fn paths(&self, key: &str) -> Option<(PathBuf, PathBuf)> { let dir = self.dir.as_ref()?; Some(( dir.join(format!("{key}.cwasm")), dir.join(format!("{key}.cwasm.sha256")), )) }
/// `Module::deserialize` trusts its bytes, so a file is loaded only when /// the sidecar hash written beside it still matches; a mismatch, a /// missing file or a deserialize error is a miss. fn load_from_disk(&self, engine: &Engine, key: &str) -> Option<Module> { let (module_path, sidecar_path) = self.paths(key)?; let bytes = std::fs::read(&module_path).ok()?; let expected = std::fs::read_to_string(&sidecar_path).ok()?; let actual = hex::encode(Sha256::digest(&bytes)); if expected.trim() != actual { tracing::warn!(path = %module_path.display(), "plugin cache file does not match its recorded hash; recompiling"); return None; } // SAFETY: the bytes were written by `write_to_disk` from // `Module::serialize` and their hash matches the sidecar that was // written with them, so they are what this process (or one with the // same engine fingerprint) produced. Wasmtime still verifies the // engine fingerprint embedded in the bytes. match unsafe { Module::deserialize(engine, &bytes) } { Ok(module) => Some(module), Err(e) => { tracing::warn!(path = %module_path.display(), error = %e, "plugin cache file could not be loaded; recompiling"); None } } }
fn write_to_disk(&self, key: &str, module: &Module) { let Some((module_path, sidecar_path)) = self.paths(key) else { return; }; let Some(dir) = self.dir.as_ref() else { return; }; let result = (|| -> std::io::Result<()> { prepare_cache_dir(dir).map_err(std::io::Error::other)?; let bytes = module .serialize() .map_err(|e| std::io::Error::other(e.to_string()))?; let sha = hex::encode(Sha256::digest(&bytes)); // Written beside the final name and renamed into place, so a // reader never sees a half-written module. let tmp = module_path.with_extension(format!("cwasm.{}.tmp", std::process::id())); std::fs::write(&tmp, &bytes)?; std::fs::rename(&tmp, &module_path)?; std::fs::write(&sidecar_path, sha)?; Ok(()) })(); if let Err(e) = result { self.warn_once(dir, "write", &e); } }
fn warn_once(&self, dir: &Path, action: &str, error: &std::io::Error) { if !self.warned.swap(true, Ordering::Relaxed) { tracing::warn!( dir = %dir.display(), error = %error, "could not {action} the plugin module cache; every plugin will be compiled on each use" ); } }}
fn tick(engine: &Engine, epoch: &AtomicU64) { engine.increment_epoch(); epoch.fetch_add(1, Ordering::Relaxed);}
impl Drop for WasmRuntime { fn drop(&mut self) { self.ticker_stop.store(true, Ordering::Relaxed); }}
impl Default for WasmRuntime { fn default() -> Self { Self::new().expect("Failed to create WASM runtime") }}
/// Linear memory a library or auth plugin may grow to. Roomy enough for a/// response at `MAX_HTTP_RESPONSE_SIZE` and the guest's own copy of it,/// and a quarter of the 4 GiB a 32-bit guest could otherwise reach.pub const LIBRARY_MEMORY_CEILING: usize = 512 * 1024 * 1024;
/// The Lua-level allocation limit a script runs under until an operator/// setting supplies one.pub const DEFAULT_SCRIPT_MEMORY_BYTES: usize = 64 * 1024 * 1024;
/// The store ceiling for a script memory limit. Twice the Lua-level limit,/// because the interpreter's own footprint and its allocator's overhead sit/// on top of the bytes Lua counts (a 64 MiB Lua limit measured at 87 MiB of/// linear memory). The Lua limit is the one that produces a message; this/// is the backstop, reached only by something the Lua limit cannot see.pub fn script_memory_ceiling(script_memory_bytes: usize) -> usize { script_memory_bytes.saturating_mul(2)}
/// The ceiling a store starts with, by plugin type.pub fn memory_ceiling(plugin_type: PluginType) -> usize { match plugin_type { PluginType::Library | PluginType::Auth => LIBRARY_MEMORY_CEILING, PluginType::Interpreter => script_memory_ceiling(DEFAULT_SCRIPT_MEMORY_BYTES), }}
/// A store's memory ceiling, and whether it has refused a growth. A refusal/// reaches the guest as `memory.grow` returning -1 and reaches the host, if/// the guest then traps, as a trap with nothing on it naming the cause;/// the flag is what tells "out of memory" from "the guest has a bug".pub struct MemoryLimiter { limits: StoreLimits, ceiling: usize, refused: Option<usize>,}
impl MemoryLimiter { pub fn new(ceiling: usize) -> Self { Self { // `memory_size` caps one linear memory, so the count is pinned // too; otherwise a module declaring several would multiply the // ceiling. One instance and one table is every plugin's shape. limits: StoreLimitsBuilder::new() .memory_size(ceiling) .memories(1) .tables(1) .instances(1) .build(), ceiling, refused: None, } }
/// Forget an earlier refusal, so one classifies only the call it /// happened in: the guest sees a refused growth as `memory.grow` /// returning -1 and may carry on. pub fn clear_refusal(&mut self) { self.refused = None; }
pub fn ceiling(&self) -> usize { self.ceiling }
/// The size the guest asked for when a growth was refused. pub fn refused(&self) -> Option<usize> { self.refused }}
impl ResourceLimiter for MemoryLimiter { fn memory_growing( &mut self, current: usize, desired: usize, maximum: Option<usize>, ) -> Result<bool> { let allowed = self.limits.memory_growing(current, desired, maximum)?; if !allowed { self.refused = Some(desired); } Ok(allowed) }
fn memory_grow_failed(&mut self, error: Error) -> Result<()> { self.limits.memory_grow_failed(error) }
fn table_growing( &mut self, current: usize, desired: usize, maximum: Option<usize>, ) -> Result<bool> { self.limits.table_growing(current, desired, maximum) }
fn table_grow_failed(&mut self, error: Error) -> Result<()> { self.limits.table_grow_failed(error) }
fn instances(&self) -> usize { self.limits.instances() }
fn tables(&self) -> usize { self.limits.tables() }
fn memories(&self) -> usize { self.limits.memories() }}
/// A store's budget of guest execution, in ticks, counting only the time/// the guest is running. Entering a host import suspends the clock and/// returning resumes it, so a script awaiting a PDS write or an HTTP fetch/// is not charged for the host's latency. The methods keep the books; the/// caller applies the returned delta with `set_epoch_deadline`, since the/// store owns this value and cannot be borrowed through it.pub struct GuestDeadline { epoch: Arc<AtomicU64>, remaining: u64, armed_at: u64, running: bool,}
impl GuestDeadline { pub fn unbounded(epoch: Arc<AtomicU64>) -> Self { Self { epoch, remaining: UNBOUNDED_TICKS, armed_at: 0, running: false, } }
/// The ticks a wall-clock budget spans. Rounded up, plus one: the epoch /// is partway through a tick whenever a budget is armed, so the first /// tick is short, and without the extra one a budget could end up to a /// tick before the seconds an operator typed. pub fn ticks_for(budget: Duration) -> u64 { budget.as_millis().div_ceil(EPOCH_TICK.as_millis()) as u64 + 1 }
fn now(&self) -> u64 { self.epoch.load(Ordering::Relaxed) }
/// Charge the ticks the guest has run since the clock last started. fn settle(&mut self) { let now = self.now(); if self.running { self.remaining = self.remaining.saturating_sub(now - self.armed_at); } self.armed_at = now; }
/// Start a fresh budget. Returns the delta for `set_epoch_deadline`. pub fn arm(&mut self, ticks: u64) -> u64 { self.remaining = ticks; self.armed_at = self.now(); self.running = true; ticks }
/// The guest has entered a host import: spend what it ran and stop /// the clock. pub fn suspend(&mut self) { self.settle(); self.running = false; }
/// The guest is about to run again. Returns the delta for /// `set_epoch_deadline`; a budget that reached zero re-arms at zero, so /// the guest is interrupted the moment it resumes rather than being /// allowed one more unbounded stretch. pub fn resume(&mut self) -> u64 { self.settle(); self.running = true; self.remaining }
pub fn remaining(&self) -> u64 { self.remaining }}
#[cfg(test)]mod tests { use super::*;
#[test] fn module_fuel_scales_the_budget_and_leaves_none_unbounded() { assert_eq!(module_fuel(Some(1_000)), 1_000 * FUEL_PER_INSTRUCTION); assert_eq!(module_fuel(None), u64::MAX); assert_eq!( module_fuel(Some(u32::MAX)), u64::from(u32::MAX) * FUEL_PER_INSTRUCTION ); }
#[test] fn test_fuel_constant_value() { // 10M fuel ≈ 100ms CPU time per spec assert_eq!(DEFAULT_FUEL, 10_000_000); }
#[test] fn test_runtime_has_fuel_enabled() { let runtime = WasmRuntime::new().expect("Failed to create runtime"); // We can verify fuel is enabled by checking we can set it on a store let mut store = wasmtime::Store::new(runtime.engine(), ()); assert!(store.set_fuel(1000).is_ok()); }
/// Calls itself `depth` times; each frame holds a return address and a /// frame pointer at the least, so 100 000 of them is well past the 512 KiB /// wasmtime allows by default. const RECURSE: &str = r#"(module (func $down (export "down") (param i32) (result i32) local.get 0 i32.eqz if (result i32) i32.const 0 else local.get 0 i32.const 1 i32.sub call $down i32.const 1 i32.add end))"#;
async fn recurse(depth: i32) -> Result<i32, wasmtime::Error> { let runtime = WasmRuntime::without_ticker().unwrap(); let module = runtime.compile(&wat::parse_str(RECURSE).unwrap()).unwrap(); let mut store = wasmtime::Store::new(runtime.engine(), ()); store.set_fuel(u64::MAX).unwrap(); store.set_epoch_deadline(UNBOUNDED_TICKS); let instance = Instance::new_async(&mut store, &module, &[]).await.unwrap(); let down = instance .get_typed_func::<i32, i32>(&mut store, "down") .unwrap(); down.call_async(&mut store, depth).await }
#[tokio::test] async fn a_guest_recursing_past_the_default_stack_still_returns() { assert_eq!(recurse(100_000).await.unwrap(), 100_000); }
/// A guest with no limit of its own still ends at the ceiling, as a trap /// the store survives rather than a fault that takes the process down. #[tokio::test] async fn unbounded_recursion_ends_in_a_stack_overflow_trap() { let error = recurse(i32::MAX).await.unwrap_err(); assert_eq!( error.downcast_ref::<Trap>(), Some(&Trap::StackOverflow), "{error:?}" ); }
#[test] fn test_compile_invalid_wasm_fails() { let runtime = WasmRuntime::new().expect("Failed to create runtime"); let result = runtime.compile(b"not valid wasm"); assert!(result.is_err()); }
/// A store left at wasmtime's default deadline traps on its first /// instruction once epoch interruption is on; the unbounded delta does /// not, and does not overflow the engine's addition either. #[test] fn unbounded_deadline_neither_traps_nor_overflows() { let runtime = WasmRuntime::without_ticker().unwrap(); let module = runtime .compile(&wat::parse_str(r#"(module (func (export "f")))"#).unwrap()) .unwrap(); for _ in 0..3 { runtime.tick(); }
let mut store = Store::new(runtime.engine(), ()); store.set_fuel(DEFAULT_FUEL).unwrap(); let instance = Instance::new(&mut store, &module, &[]).unwrap(); let f = instance.get_typed_func::<(), ()>(&mut store, "f").unwrap(); let err = f.call(&mut store, ()).unwrap_err(); assert_eq!(err.downcast_ref::<Trap>(), Some(&Trap::Interrupt));
let mut store = Store::new(runtime.engine(), ()); store.set_fuel(DEFAULT_FUEL).unwrap(); store.set_epoch_deadline(UNBOUNDED_TICKS); let instance = Instance::new(&mut store, &module, &[]).unwrap(); let f = instance.get_typed_func::<(), ()>(&mut store, "f").unwrap(); f.call(&mut store, ()).unwrap(); }
/// The mirror and the engine's epoch move together: a deadline of a /// thousand ticks holds through the 999th tick and trips on the /// thousandth, exactly where the mirror says it should. #[test] fn the_mirror_and_the_engine_epoch_do_not_drift_over_a_thousand_ticks() { let runtime = WasmRuntime::without_ticker().unwrap(); let module = runtime .compile(&wat::parse_str(r#"(module (func (export "f")))"#).unwrap()) .unwrap(); let mut store = Store::new(runtime.engine(), ()); store.set_fuel(DEFAULT_FUEL).unwrap(); store.set_epoch_deadline(1000); let instance = Instance::new(&mut store, &module, &[]).unwrap(); let f = instance.get_typed_func::<(), ()>(&mut store, "f").unwrap();
let start = runtime.epoch().load(Ordering::Relaxed); for _ in 0..999 { runtime.tick(); } assert_eq!(runtime.epoch().load(Ordering::Relaxed) - start, 999); f.call(&mut store, ()).unwrap(); runtime.tick(); assert_eq!(runtime.epoch().load(Ordering::Relaxed) - start, 1000); let err = f.call(&mut store, ()).unwrap_err(); assert_eq!(err.downcast_ref::<Trap>(), Some(&Trap::Interrupt)); }
#[test] fn the_ticker_advances_the_mirror() { let runtime = WasmRuntime::new().unwrap(); let start = runtime.epoch().load(Ordering::Relaxed); std::thread::sleep(EPOCH_TICK * 5); assert!(runtime.epoch().load(Ordering::Relaxed) > start); }
#[test] fn ticks_round_a_budget_up() { assert_eq!(GuestDeadline::ticks_for(Duration::from_secs(1)), 101); assert_eq!(GuestDeadline::ticks_for(Duration::from_millis(15)), 3); assert_eq!(GuestDeadline::ticks_for(Duration::ZERO), 1); }
/// Ticks that pass while suspended are not charged; ticks that pass /// while running are, and the budget re-arms with what is left. #[test] fn a_suspended_deadline_does_not_spend_host_time() { let epoch = Arc::new(AtomicU64::new(0)); let mut deadline = GuestDeadline::unbounded(epoch.clone()); assert_eq!(deadline.arm(10), 10);
epoch.fetch_add(3, Ordering::Relaxed); deadline.suspend(); assert_eq!(deadline.remaining(), 7);
epoch.fetch_add(100, Ordering::Relaxed); assert_eq!(deadline.resume(), 7); assert_eq!(deadline.remaining(), 7);
epoch.fetch_add(2, Ordering::Relaxed); deadline.suspend(); deadline.suspend(); assert_eq!(deadline.remaining(), 5);
epoch.fetch_add(9, Ordering::Relaxed); assert_eq!(deadline.resume(), 5); epoch.fetch_add(9, Ordering::Relaxed); assert_eq!(deadline.resume(), 0); assert_eq!(deadline.resume(), 0); }
#[test] fn fuel_by_plugin_type() { assert_eq!(fuel_for(PluginType::Library), DEFAULT_FUEL); assert_eq!(fuel_for(PluginType::Auth), DEFAULT_FUEL); assert_eq!(fuel_for(PluginType::Interpreter), u64::MAX); }
#[test] fn ceilings_by_plugin_type() { assert_eq!(memory_ceiling(PluginType::Library), 512 * 1024 * 1024); assert_eq!(memory_ceiling(PluginType::Auth), 512 * 1024 * 1024); assert_eq!(memory_ceiling(PluginType::Interpreter), 128 * 1024 * 1024); assert_eq!(script_memory_ceiling(64 * 1024 * 1024), 128 * 1024 * 1024); }
/// One memory per store, so the ceiling is the store's and not each /// memory's. #[test] fn the_limiter_pins_one_memory_one_table_one_instance() { let limiter = MemoryLimiter::new(1024 * 1024); assert_eq!(limiter.memories(), 1); assert_eq!(limiter.tables(), 1); assert_eq!(limiter.instances(), 1); }
/// The limiter records what it refused, and forgets it on request. #[test] fn the_limiter_records_a_refusal() { let mut limiter = MemoryLimiter::new(1024 * 1024); assert!(limiter.memory_growing(0, 512 * 1024, None).unwrap()); assert_eq!(limiter.refused(), None); assert!( !limiter .memory_growing(512 * 1024, 2 * 1024 * 1024, None) .unwrap() ); assert_eq!(limiter.refused(), Some(2 * 1024 * 1024)); assert_eq!(limiter.ceiling(), 1024 * 1024); limiter.clear_refusal(); assert_eq!(limiter.refused(), None); }
fn plugin(id: &str, wat: &str) -> Arc<LoadedPlugin> { let manifest: crate::plugin::PluginManifest = serde_json::from_value(serde_json::json!({ "id": id, "name": id, "version": "1.0.0", "api_version": "2", "plugin_type": "library", "namespace": id, "capabilities": [], })) .unwrap(); Arc::new(LoadedPlugin { info: manifest.clone().into(), source: crate::plugin::PluginSource::File { path: id.into() }, wasm_bytes: wat::parse_str(wat).unwrap(), manifest: Some(manifest), }) }
fn answering(value: i32) -> String { format!(r#"(module (func (export "answer") (result i32) i32.const {value}))"#) }
fn answer(runtime: &WasmRuntime, module: &Module) -> i32 { let mut store = Store::new(runtime.engine(), ()); store.set_fuel(DEFAULT_FUEL).unwrap(); store.set_epoch_deadline(UNBOUNDED_TICKS); let instance = Instance::new(&mut store, module, &[]).unwrap(); instance .get_typed_func::<(), i32>(&mut store, "answer") .unwrap() .call(&mut store, ()) .unwrap() }
/// A directory of this test's own under the system temp directory. struct CacheDir(PathBuf);
impl CacheDir { fn new(tag: &str) -> Self { let dir = std::env::temp_dir().join(format!( "happyview-module-cache-test-{tag}-{}-{}", std::process::id(), Instant::now().elapsed().as_nanos() )); let _ = std::fs::remove_dir_all(&dir); Self(dir) }
fn files(&self) -> Vec<String> { let mut names: Vec<String> = std::fs::read_dir(&self.0) .map(|entries| { entries .flatten() .map(|e| e.file_name().to_string_lossy().into_owned()) .collect() }) .unwrap_or_default(); names.sort(); names } }
impl Drop for CacheDir { fn drop(&mut self) { let _ = std::fs::remove_dir_all(&self.0); let _ = std::fs::remove_file(&self.0); } }
#[test] fn a_second_lookup_of_the_same_plugin_does_not_recompile() { let runtime = WasmRuntime::without_ticker().unwrap(); let plugin = plugin("p", &answering(7)); let first = runtime.module_for(&plugin).unwrap(); let second = runtime.module_for(&plugin).unwrap(); assert_eq!(runtime.compile_count(), 1); assert_eq!(answer(&runtime, &first), 7); assert_eq!(answer(&runtime, &second), 7); }
/// The same bytes under a fresh registration hit by hash; different /// bytes under the same id miss and replace the entry. #[test] fn a_re_registration_hits_by_hash_and_new_bytes_replace_the_module() { let runtime = WasmRuntime::without_ticker().unwrap(); runtime.module_for(&plugin("p", &answering(1))).unwrap(); runtime.module_for(&plugin("p", &answering(1))).unwrap(); assert_eq!(runtime.compile_count(), 1); let module = runtime.module_for(&plugin("p", &answering(2))).unwrap(); assert_eq!(runtime.compile_count(), 2); assert_eq!(answer(&runtime, &module), 2); }
#[test] fn a_disk_entry_written_by_one_runtime_is_used_by_a_fresh_one() { let dir = CacheDir::new("disk"); let plugin = plugin("p", &answering(9));
let first = WasmRuntime::with_cache_dir(Some(dir.0.clone())).unwrap(); first.module_for(&plugin).unwrap(); assert_eq!(first.compile_count(), 1); let key = first.cache_key(&hex::encode(Sha256::digest(&plugin.wasm_bytes))); assert_eq!( dir.files(), vec![format!("{key}.cwasm"), format!("{key}.cwasm.sha256")] );
let second = WasmRuntime::with_cache_dir(Some(dir.0.clone())).unwrap(); let module = second.module_for(&plugin).unwrap(); assert_eq!(second.compile_count(), 0); assert_eq!(answer(&second, &module), 9); }
#[test] fn a_cache_file_whose_sidecar_does_not_match_is_recompiled_and_rewritten() { let dir = CacheDir::new("sidecar"); let plugin = plugin("p", &answering(3)); let first = WasmRuntime::with_cache_dir(Some(dir.0.clone())).unwrap(); first.module_for(&plugin).unwrap(); let key = first.cache_key(&hex::encode(Sha256::digest(&plugin.wasm_bytes))); let sidecar = dir.0.join(format!("{key}.cwasm.sha256")); std::fs::write(&sidecar, "0000").unwrap();
let second = WasmRuntime::with_cache_dir(Some(dir.0.clone())).unwrap(); let module = second.module_for(&plugin).unwrap(); assert_eq!(second.compile_count(), 1); assert_eq!(answer(&second, &module), 3); assert_ne!(std::fs::read_to_string(&sidecar).unwrap(), "0000");
// Corrupt bytes with a matching sidecar are a miss too. let module_path = dir.0.join(format!("{key}.cwasm")); std::fs::write(&module_path, b"not a module").unwrap(); std::fs::write(&sidecar, hex::encode(Sha256::digest(b"not a module"))).unwrap(); let third = WasmRuntime::with_cache_dir(Some(dir.0.clone())).unwrap(); assert_eq!(answer(&third, &third.module_for(&plugin).unwrap()), 3); assert_eq!(third.compile_count(), 1); }
/// The key carries the engine fingerprint, so a module serialized by /// another engine lives under another name and is never opened. #[test] fn a_key_under_a_different_engine_fingerprint_is_ignored() { let dir = CacheDir::new("engine"); let plugin = plugin("p", &answering(4)); let runtime = WasmRuntime::with_cache_dir(Some(dir.0.clone())).unwrap(); let sha = hex::encode(Sha256::digest(&plugin.wasm_bytes)); let key = runtime.cache_key(&sha); assert!(key.starts_with(&format!("{sha}-"))); assert_eq!(key.len(), sha.len() + 1 + 16);
let module = runtime.compile(&plugin.wasm_bytes).unwrap(); let bytes = module.serialize().unwrap(); std::fs::create_dir_all(&dir.0).unwrap(); let foreign = format!("{sha}-{:016x}", 0xdead_beef_u64); std::fs::write(dir.0.join(format!("{foreign}.cwasm")), &bytes).unwrap(); std::fs::write( dir.0.join(format!("{foreign}.cwasm.sha256")), hex::encode(Sha256::digest(&bytes)), ) .unwrap();
let fresh = WasmRuntime::with_cache_dir(Some(dir.0.clone())).unwrap(); fresh.module_for(&plugin).unwrap(); assert_eq!(fresh.compile_count(), 1); }
#[test] fn an_unwritable_cache_directory_warns_once_and_every_call_succeeds() { let dir = CacheDir::new("unwritable"); // A file where the directory should be: nothing under it can be // created. std::fs::write(&dir.0, b"").unwrap(); let runtime = WasmRuntime::with_cache_dir(Some(dir.0.clone())).unwrap(); for value in [1, 2, 3] { let module = runtime.module_for(&plugin("p", &answering(value))).unwrap(); assert_eq!(answer(&runtime, &module), value); } assert_eq!(runtime.compile_count(), 3); assert!(runtime.modules.warned.load(Ordering::Relaxed)); }
/// A fresh directory is created for this user alone. #[cfg(unix)] #[test] fn a_fresh_cache_directory_is_created_private() { use std::os::unix::fs::MetadataExt; let dir = CacheDir::new("fresh"); let runtime = WasmRuntime::with_cache_dir(Some(dir.0.clone())).unwrap(); assert!(runtime.modules.dir.is_some()); let metadata = std::fs::metadata(&dir.0).unwrap(); assert_eq!(metadata.mode() & 0o777, 0o700); assert_eq!(metadata.uid(), unsafe { libc::geteuid() }); assert!(!runtime.modules.warned.load(Ordering::Relaxed)); }
/// A directory another user can enter is refused: the cache stays in /// memory, one warning is logged, and nothing is written there. #[cfg(unix)] #[test] fn a_cache_directory_with_group_or_other_bits_is_refused() { use std::os::unix::fs::PermissionsExt; let dir = CacheDir::new("shared"); std::fs::create_dir_all(&dir.0).unwrap(); std::fs::set_permissions(&dir.0, std::fs::Permissions::from_mode(0o755)).unwrap(); let runtime = WasmRuntime::with_cache_dir(Some(dir.0.clone())).unwrap(); assert!(runtime.modules.dir.is_none()); assert!(runtime.modules.warned.load(Ordering::Relaxed)); let module = runtime.module_for(&plugin("p", &answering(5))).unwrap(); assert_eq!(answer(&runtime, &module), 5); assert!(dir.files().is_empty()); }
/// Ownership by any other uid is refused even at mode 0700; the check /// takes the uid as an argument because no test can create a directory /// owned by someone else. #[cfg(unix)] #[test] fn a_cache_directory_owned_by_another_user_is_refused() { let dir = CacheDir::new("owner"); WasmRuntime::with_cache_dir(Some(dir.0.clone())).unwrap(); let metadata = std::fs::metadata(&dir.0).unwrap(); let me = unsafe { libc::geteuid() }; assert!(directory_is_private(&metadata, me).is_ok()); let err = directory_is_private(&metadata, me.wrapping_add(1)).unwrap_err(); assert!(err.contains("owned by uid"), "{err}"); }
#[test] fn the_sweep_deletes_orphans_and_keeps_live_modules_and_other_files() { let dir = CacheDir::new("sweep"); let live = plugin("live", &answering(1)); let gone = plugin("gone", &answering(2)); let runtime = WasmRuntime::with_cache_dir(Some(dir.0.clone())).unwrap(); runtime.module_for(&live).unwrap(); runtime.module_for(&gone).unwrap(); let live_sha = hex::encode(Sha256::digest(&live.wasm_bytes)); let live_key = runtime.cache_key(&live_sha); std::fs::write(dir.0.join("notes.txt"), b"keep").unwrap(); std::fs::write( dir.0.join(format!("{live_sha}-0000000000000000.cwasm")), b"x", ) .unwrap(); assert_eq!(dir.files().len(), 6);
runtime.sweep_cache_dir(&HashSet::from([live_sha])); assert_eq!( dir.files(), vec![ format!("{live_key}.cwasm"), format!("{live_key}.cwasm.sha256"), "notes.txt".to_string(), ] ); }
#[test] fn an_unbounded_deadline_stays_unbounded_across_suspension() { let epoch = Arc::new(AtomicU64::new(0)); let mut deadline = GuestDeadline::unbounded(epoch.clone()); epoch.fetch_add(1_000_000, Ordering::Relaxed); deadline.suspend(); assert_eq!(deadline.resume(), UNBOUNDED_TICKS); epoch.fetch_add(1_000_000, Ordering::Relaxed); deadline.suspend(); assert!(deadline.resume() > UNBOUNDED_TICKS - 2_000_000); }}