1#![cfg(not(target_arch = "wasm32"))]
2
3use std::cell::{Cell, RefCell};
4use std::collections::{HashMap, HashSet, VecDeque};
5use std::rc::Rc;
6use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
7
8use wasmtime::{
9 Caller, Config, Engine, Extern, Func, Instance, Linker, Memory, Module, Store, StoreLimits,
10 StoreLimitsBuilder, Val, ValType,
11};
12
13use crate::core::{Promise, PromiseState, Value};
14use crate::extension::{ExtensionExport, ExtensionManifest, WasmAbi, WasmExtensionProvider};
15use crate::hta;
16use crate::wasm_binding::{MemoryBindingPlan, WasmtimeMemoryExecutor};
17
18struct Session {
19 store: Store<StoreLimits>,
20 instance: Instance,
21}
22
23#[derive(Clone)]
26pub struct CompiledWasmModule {
27 engine: Engine,
28 module: Module,
29 exports: Vec<(String, ExtensionExport)>,
30}
31
32impl CompiledWasmModule {
33 pub fn compile(bytes: &[u8]) -> Result<Self, String> {
34 let exports = crate::direct_wasm::exports(bytes)?;
35 let mut config = Config::new();
36 config.consume_fuel(true);
37 let engine = Engine::new(&config)
38 .map_err(|error| format!("extension/engine-unavailable: {error}"))?;
39 let module = Module::new(&engine, bytes)
40 .map_err(|error| format!("extension/module-invalid: {error}"))?;
41 if module.imports().next().is_some() {
42 return Err("extension/module-invalid: extension modules must be import-free".into());
43 }
44 Ok(Self {
45 engine,
46 module,
47 exports,
48 })
49 }
50
51 pub fn provider(&self) -> WasmtimeExtensionProvider {
52 WasmtimeExtensionProvider {
53 mode: ProviderMode::Direct {
54 engine: self.engine.clone(),
55 module: self.module.clone(),
56 session: RefCell::new(None),
57 },
58 }
59 }
60
61 pub fn direct_exports(&self) -> Result<Vec<(String, ExtensionExport)>, String> {
62 Ok(self.exports.clone())
63 }
64}
65
66pub struct WasmtimeExtensionProvider {
68 mode: ProviderMode,
69}
70
71enum ProviderMode {
72 Direct {
73 engine: Engine,
74 module: Module,
75 session: RefCell<Option<Session>>,
76 },
77 Memory(WasmtimeMemoryExecutor),
78 Hta(Rc<HtaProviderState>),
79}
80
81impl WasmtimeExtensionProvider {
82 pub fn compile(bytes: &[u8]) -> Result<Self, String> {
83 Ok(CompiledWasmModule::compile(bytes)?.provider())
84 }
85
86 pub fn compile_memory(bytes: &[u8], plan: MemoryBindingPlan) -> Result<Self, String> {
87 Ok(Self {
88 mode: ProviderMode::Memory(WasmtimeMemoryExecutor::compile(bytes, plan)?),
89 })
90 }
91
92 pub fn compile_hta(bytes: &[u8]) -> Result<Self, String> {
93 Self::compile_hta_with_host_handler(bytes, None)
94 }
95
96 pub fn drain_lifecycle_events(&self) -> Vec<HtaProviderEvent> {
97 match &self.mode {
98 ProviderMode::Hta(state) => state.trace.drain(),
99 _ => Vec::new(),
100 }
101 }
102
103 pub fn compile_hta_with_host_handler(
104 bytes: &[u8],
105 host_handler: Option<Rc<dyn Fn(String, String, Vec<Value>) -> Result<Value, String>>>,
106 ) -> Result<Self, String> {
107 Self::compile_hta_parts(bytes, None, host_handler)
108 }
109
110 pub fn compile_hta_with_library(
111 bytes: &[u8],
112 library_bytes: &[u8],
113 host_handler: Option<Rc<dyn Fn(String, String, Vec<Value>) -> Result<Value, String>>>,
114 ) -> Result<Self, String> {
115 Self::compile_hta_parts(bytes, Some(library_bytes), host_handler)
116 }
117
118 fn compile_hta_parts(
119 bytes: &[u8],
120 library_bytes: Option<&[u8]>,
121 host_handler: Option<Rc<dyn Fn(String, String, Vec<Value>) -> Result<Value, String>>>,
122 ) -> Result<Self, String> {
123 let (engine, module) = compile_hta_module(bytes, library_bytes.is_some())?;
124 let library = library_bytes
125 .map(|bytes| {
126 Module::new(&engine, bytes)
127 .map_err(|error| format!("extension/module-invalid: {error}"))
128 })
129 .transpose()?;
130 Ok(Self {
131 mode: ProviderMode::Hta(Rc::new(HtaProviderState {
132 engine,
133 module,
134 library,
135 session: RefCell::new(None),
136 host_handler,
137 timeout: hta_timeout(),
138 trace: HtaProviderTrace::new(),
139 })),
140 })
141 }
142}
143
144impl WasmExtensionProvider for WasmtimeExtensionProvider {
145 fn supports(&self, abi: WasmAbi) -> bool {
146 matches!(
147 (&self.mode, abi),
148 (ProviderMode::Direct { .. }, WasmAbi::CoreV1)
149 | (ProviderMode::Memory(_), WasmAbi::MemoryV1)
150 | (ProviderMode::Hta(_), WasmAbi::HtaV1)
151 )
152 }
153
154 fn capabilities(&self) -> Vec<String> {
155 if matches!(&self.mode, ProviderMode::Hta(_)) {
156 return hta_capabilities();
157 }
158 Vec::new()
159 }
160
161 fn start(&self, manifest: &ExtensionManifest) -> Result<(), String> {
162 if let ProviderMode::Hta(state) = &self.mode {
163 return state.start(manifest);
164 }
165 if !manifest.capabilities.is_empty() || !manifest.host_call_capabilities.is_empty() {
166 return Err(format!(
167 "extension/capability-denied: {:?} for {}",
168 manifest.capabilities, manifest.namespace
169 ));
170 }
171 if let ProviderMode::Memory(executor) = &self.mode {
172 let plan = executor.plan();
173 if manifest.exports.len() != plan.functions.len()
174 || manifest.exports.iter().any(|(name, specification)| {
175 plan.functions
176 .iter()
177 .find(|function| function.name == *name)
178 .map_or(true, |function| {
179 specification.raw_name(name) != function.wasm_export
180 })
181 })
182 {
183 return Err(format!(
184 "extension/manifest-mismatch: memory.v1 exports for {} do not match bindings.edn",
185 manifest.namespace
186 ));
187 }
188 return Ok(());
189 }
190 let ProviderMode::Direct {
191 engine,
192 module,
193 session,
194 } = &self.mode
195 else {
196 unreachable!()
197 };
198 let limits = StoreLimitsBuilder::new()
199 .memory_size(64 * 1024 * 1024)
200 .instances(1)
201 .memories(1)
202 .tables(1)
203 .build();
204 let mut store = Store::new(engine, limits);
205 store.limiter(|limits| limits);
206 let instance = Instance::new(&mut store, module, &[])
207 .map_err(|error| format!("extension/module-invalid: {error}"))?;
208 for (name, specification) in &manifest.exports {
209 let raw_name = specification.raw_name(name);
210 let function = instance.get_func(&mut store, raw_name).ok_or_else(|| {
211 format!(
212 "extension/malformed: module has no export {raw_name} for public name {name}"
213 )
214 })?;
215 if function.ty(&store).results().len() > 1 {
216 return Err(format!(
217 "extension/abi-type-unsupported: {name} has multiple results"
218 ));
219 }
220 }
221 *session.borrow_mut() = Some(Session { store, instance });
222 Ok(())
223 }
224
225 fn invoke(
226 &self,
227 manifest: &ExtensionManifest,
228 export: &str,
229 arguments: &[Value],
230 ) -> Result<Value, String> {
231 if let ProviderMode::Memory(executor) = &self.mode {
232 return executor.invoke(export, arguments);
233 }
234 if let ProviderMode::Hta(state) = &self.mode {
235 return state.invoke(manifest, export, arguments);
236 }
237 let ProviderMode::Direct { session, .. } = &self.mode else {
238 unreachable!()
239 };
240 let specification = manifest
241 .exports
242 .iter()
243 .find(|(name, _)| name == export)
244 .map(|(_, specification)| specification)
245 .ok_or_else(|| format!("extension/export-missing: {export}"))?;
246 let raw_name = specification.raw_name(export);
247 let mut session = session.borrow_mut();
248 let session = session
249 .as_mut()
250 .ok_or_else(|| format!("extension/not-started: {}", manifest.namespace))?;
251 let function = session
252 .instance
253 .get_func(&mut session.store, raw_name)
254 .ok_or_else(|| format!("extension/export-missing: {export} -> {raw_name}"))?;
255 let values = specification
256 .arguments
257 .iter()
258 .zip(arguments)
259 .map(|(wire_type, value)| argument(export, wire_type, value))
260 .collect::<Result<Vec<_>, _>>()?;
261 let mut results = if specification.returns == "void" {
262 Vec::new()
263 } else {
264 vec![default_result(&specification.returns)?]
265 };
266 session
267 .store
268 .set_fuel(10_000_000)
269 .map_err(|error| format!("extension/execution-limit: {error}"))?;
270 function
271 .call(&mut session.store, &values, &mut results)
272 .map_err(|error| {
273 format!(
274 "extension/invoke-failed: {}/{} ({error})",
275 manifest.namespace, export
276 )
277 })?;
278 result(export, &specification.returns, results.into_iter().next())
279 }
280
281 fn cancel(&self, _manifest: &ExtensionManifest, _request: u64) -> Result<(), String> {
282 if let ProviderMode::Hta(state) = &self.mode {
283 return state.cancel(_request);
284 }
285 Err("extension/cancel-unsupported: core.v1 calls are synchronous".into())
286 }
287
288 fn release(&self, manifest: &ExtensionManifest, handle: &Value) -> Result<(), String> {
289 if let ProviderMode::Hta(state) = &self.mode {
290 return state.release(manifest, handle);
291 }
292 Err("extension/release-unsupported: provider has no HTA handle boundary".into())
293 }
294
295 fn shutdown(&self, manifest: &ExtensionManifest) {
296 match &self.mode {
297 ProviderMode::Direct { session, .. } => {
298 session.borrow_mut().take();
299 }
300 ProviderMode::Hta(state) => state.shutdown(manifest),
301 ProviderMode::Memory(_) => {}
302 }
303 }
304}
305
306const MAX_HTA_FRAME_BYTES: usize = 64 * 1024 * 1024;
307const DEFAULT_HTA_TIMEOUT: Duration = Duration::from_secs(120);
308
309pub const HTA_PROVIDER_EVENT_SCHEMA: &str = "hara.hta.provider.event/0-alpha";
310
311#[derive(Debug, Clone, PartialEq, Eq)]
312pub struct HtaProviderEvent {
313 pub schema: &'static str,
314 pub sequence: u64,
315 pub origin: &'static str,
316 pub event: &'static str,
317 pub request: Option<u64>,
318 pub operation: Option<String>,
319 pub status: Option<String>,
320 pub code: Option<String>,
321}
322
323struct HtaProviderTrace {
324 sequence: Cell<u64>,
325 shutdown: Cell<bool>,
326 events: RefCell<Vec<HtaProviderEvent>>,
327}
328
329impl HtaProviderTrace {
330 fn new() -> Self {
331 Self {
332 sequence: Cell::new(0),
333 shutdown: Cell::new(false),
334 events: RefCell::new(Vec::new()),
335 }
336 }
337
338 fn emit(
339 &self,
340 event: &'static str,
341 request: Option<u64>,
342 operation: Option<String>,
343 status: Option<&str>,
344 code: Option<String>,
345 ) {
346 let sequence = self.sequence.get() + 1;
347 self.sequence.set(sequence);
348 self.events.borrow_mut().push(HtaProviderEvent {
349 schema: HTA_PROVIDER_EVENT_SCHEMA,
350 sequence,
351 origin: "wasmtime",
352 event,
353 operation,
354 request,
355 status: status.map(str::to_owned),
356 code,
357 });
358 }
359
360 fn emit_shutdown(&self, status: Option<&str>, code: Option<String>) {
361 if self.shutdown.replace(true) {
362 return;
363 }
364 self.emit("shutdown", None, None, status, code);
365 }
366
367 fn drain(&self) -> Vec<HtaProviderEvent> {
368 std::mem::take(&mut *self.events.borrow_mut())
369 }
370}
371
372fn hta_capabilities() -> Vec<String> {
373 std::env::var("HARA_HTA_CAPABILITIES")
374 .unwrap_or_default()
375 .split([',', ' ', '\n', '\t'])
376 .filter(|capability| !capability.is_empty())
377 .map(str::to_owned)
378 .collect()
379}
380
381struct HtaPending {
382 promise: Promise,
383 deadline: Option<Instant>,
384 operation: String,
385}
386
387struct HtaSession {
388 store: Store<StoreLimits>,
389 memory: Memory,
390 allocator: Func,
391 deallocator: Func,
392 start: Func,
393 next_event: Func,
394 deliver: Func,
395 cancel: Func,
396 drop_task: Func,
397 release: Func,
398 pending: HashMap<u64, HtaPending>,
399 host_promises: HashMap<u64, Promise>,
400 host_calls_seen: HashSet<u64>,
401 handles: HashSet<(String, String, u64)>,
402 deliveries: VecDeque<(u64, bool, Value)>,
403}
404
405struct HtaProviderState {
406 engine: Engine,
407 module: Module,
408 library: Option<Module>,
409 session: RefCell<Option<HtaSession>>,
410 host_handler: Option<Rc<dyn Fn(String, String, Vec<Value>) -> Result<Value, String>>>,
411 timeout: Option<Duration>,
412 trace: HtaProviderTrace,
413}
414
415impl HtaProviderState {
416 fn start(&self, manifest: &ExtensionManifest) -> Result<(), String> {
417 if manifest.provider != "wasm" || manifest.abi != WasmAbi::HtaV1 {
418 return Err(
419 "extension/manifest-mismatch: HTA Wasm provider requires :wasm/:hta.v1".into(),
420 );
421 }
422 let capabilities = hta_capabilities();
423 if manifest
424 .capabilities
425 .iter()
426 .chain(manifest.host_call_capabilities.values().flatten())
427 .any(|capability| !capabilities.contains(capability))
428 {
429 return Err(format!(
430 "extension/capability-denied: {:?} for {}",
431 manifest.capabilities, manifest.namespace
432 ));
433 }
434 if !manifest.host_calls.is_empty() && self.host_handler.is_none() {
435 return Err(format!(
436 "extension/host-unavailable: {} declares host calls",
437 manifest.namespace
438 ));
439 }
440 if self.session.borrow().is_some() {
441 return Err(format!(
442 "extension/start: session already exists for {}",
443 manifest.namespace
444 ));
445 }
446
447 let mut linker = Linker::new(&self.engine);
448 linker
449 .func_wrap(
450 "env",
451 "hara_random_fill",
452 |mut caller: Caller<'_, StoreLimits>, pointer: i32, length: i32| -> i32 {
453 if pointer < 0 || length < 0 {
454 return 1;
455 }
456 let Some(Extern::Memory(memory)) = caller.get_export("memory") else {
457 return 1;
458 };
459 let mut bytes = vec![0_u8; length as usize];
460 if getrandom::getrandom(&mut bytes).is_err()
461 || memory.write(&mut caller, pointer as usize, &bytes).is_err()
462 {
463 return 1;
464 }
465 0
466 },
467 )
468 .map_err(|error| format!("extension/engine-unavailable: {error}"))?;
469 linker
470 .func_wrap(
471 "env",
472 "hara_time_ms",
473 |_caller: Caller<'_, StoreLimits>| -> i64 {
474 SystemTime::now()
475 .duration_since(UNIX_EPOCH)
476 .map(|value| value.as_millis() as i64)
477 .unwrap_or_default()
478 },
479 )
480 .map_err(|error| format!("extension/engine-unavailable: {error}"))?;
481 linker
482 .func_wrap(
483 "env",
484 "hara_time_ns",
485 |_caller: Caller<'_, StoreLimits>| -> i64 {
486 SystemTime::now()
487 .duration_since(UNIX_EPOCH)
488 .map(|value| value.as_nanos() as i64)
489 .unwrap_or_default()
490 },
491 )
492 .map_err(|error| format!("extension/engine-unavailable: {error}"))?;
493
494 let limits = StoreLimitsBuilder::new()
495 .memory_size(64 * 1024 * 1024)
496 .instances(if self.library.is_some() { 2 } else { 1 })
497 .memories(1)
498 .tables(1)
499 .build();
500 let mut store = Store::new(&self.engine, limits);
501 store.limiter(|limits| limits);
502 if let Some(library) = &self.library {
503 let library_instance = Instance::new(&mut store, library, &[]).map_err(|error| {
504 format!("extension/module-invalid: wrapped library cannot instantiate: {error}")
505 })?;
506 for import in self.module.imports() {
507 if import.module() != "hara/library" {
508 continue;
509 }
510 let function = library_instance
511 .get_func(&mut store, import.name())
512 .ok_or_else(|| {
513 format!(
514 "extension/module-invalid: wrapped library has no export {}",
515 import.name()
516 )
517 })?;
518 linker
519 .define(&mut store, import.module(), import.name(), function)
520 .map_err(|error| format!("extension/module-invalid: {error}"))?;
521 }
522 }
523 let instance = linker
524 .instantiate(&mut store, &self.module)
525 .map_err(|error| format!("extension/module-invalid: {error}"))?;
526 let memory = instance
527 .get_memory(&mut store, "memory")
528 .ok_or_else(|| "extension/malformed: module has no export memory".to_owned())?;
529 let allocator = require_export(&instance, &mut store, "hta_alloc")?;
530 let deallocator = require_export(&instance, &mut store, "hta_dealloc")?;
531 let abi_version = require_export(&instance, &mut store, "hta_abi_version")?;
532 let start = require_export(&instance, &mut store, "hta_start")?;
533 let next_event = require_export(&instance, &mut store, "hta_next_event")?;
534 let deliver = require_export(&instance, &mut store, "hta_deliver")?;
535 let cancel = require_export(&instance, &mut store, "hta_cancel")?;
536 let drop_task = require_export(&instance, &mut store, "hta_drop_task")?;
537 expect_signature(
538 &mut store,
539 &allocator,
540 &[ValType::I32],
541 &[ValType::I32],
542 "hta_alloc",
543 )?;
544 expect_signature(
545 &mut store,
546 &deallocator,
547 &[ValType::I32, ValType::I32],
548 &[],
549 "hta_dealloc",
550 )?;
551 expect_signature(
552 &mut store,
553 &abi_version,
554 &[],
555 &[ValType::I32],
556 "hta_abi_version",
557 )?;
558 expect_signature(
559 &mut store,
560 &start,
561 &[ValType::I32, ValType::I32],
562 &[ValType::I64],
563 "hta_start",
564 )?;
565 expect_signature(
566 &mut store,
567 &next_event,
568 &[],
569 &[ValType::I64],
570 "hta_next_event",
571 )?;
572 expect_signature(
573 &mut store,
574 &deliver,
575 &[ValType::I32, ValType::I32],
576 &[ValType::I32],
577 "hta_deliver",
578 )?;
579 expect_signature(
580 &mut store,
581 &cancel,
582 &[ValType::I64],
583 &[ValType::I32],
584 "hta_cancel",
585 )?;
586 expect_signature(
587 &mut store,
588 &drop_task,
589 &[ValType::I64],
590 &[ValType::I32],
591 "hta_drop_task",
592 )?;
593 let release = require_export(&instance, &mut store, "hta_release")?;
594 expect_signature(
595 &mut store,
596 &release,
597 &[ValType::I32, ValType::I32],
598 &[ValType::I32],
599 "hta_release",
600 )?;
601 let version = call_i32(&mut store, &abi_version, &[], "hta_abi_version")?;
602 if !(1..=4).contains(&version) {
603 return Err(format!(
604 "extension/abi-version-unsupported: {}",
605 manifest.namespace
606 ));
607 }
608 *self.session.borrow_mut() = Some(HtaSession {
609 store,
610 memory,
611 allocator,
612 deallocator,
613 start,
614 next_event,
615 deliver,
616 cancel,
617 drop_task,
618 release,
619 pending: HashMap::new(),
620 host_promises: HashMap::new(),
621 host_calls_seen: HashSet::new(),
622 handles: HashSet::new(),
623 deliveries: VecDeque::new(),
624 });
625 self.trace.emit("start", None, None, Some("ok"), None);
626 Ok(())
627 }
628
629 fn invoke(
630 self: &Rc<Self>,
631 manifest: &ExtensionManifest,
632 export: &str,
633 arguments: &[Value],
634 ) -> Result<Value, String> {
635 let promise = Promise::new();
636 let (task, operation) = {
637 let mut session_ref = self.session.borrow_mut();
638 let session = session_ref
639 .as_mut()
640 .ok_or_else(|| "hta/session-closed".to_owned())?;
641 let operation = manifest
642 .operations
643 .get(export)
644 .cloned()
645 .unwrap_or_else(|| export.to_owned());
646 let arguments_value = Value::Vector(arguments.to_vec().into());
647 validate_handles_in_value(&arguments_value, manifest)?;
648 validate_live_handles(&arguments_value, &session.handles)?;
649 let request = hta::encode(&Value::Vector(
650 vec![
651 Value::String(operation.clone()),
652 Value::Vector(arguments.to_vec().into()),
653 ]
654 .into(),
655 ))?;
656 let task = execute_start(session, &request)?;
657 if task <= 0 {
658 return Err(format!("hta/start-failed: {}", manifest.namespace));
659 }
660 if session.pending.contains_key(&(task as u64)) {
661 let _ = cancel_task_on_session(session, task as u64);
662 let _ = drop_task_on_session(session, task as u64);
663 return Err(format!("hta/task-duplicate: {}", task));
664 }
665 session.pending.insert(
666 task as u64,
667 HtaPending {
668 promise: promise.clone(),
669 deadline: self.timeout.map(|timeout| Instant::now() + timeout),
670 operation: operation.clone(),
671 },
672 );
673 (task as u64, operation)
674 };
675 self.trace
676 .emit("call-enter", Some(task), Some(operation), None, None);
677 let weak = Rc::downgrade(self);
678 let manifest_for_poll = manifest.clone();
679 promise.set_poller(Rc::new(move || {
680 if let Some(state) = weak.upgrade() {
681 if let Err(error) = state.pump(&manifest_for_poll) {
682 state.fail_all(error);
683 }
684 }
685 }));
686 let weak = Rc::downgrade(self);
687 let manifest_for_wait = manifest.clone();
688 let waiting = promise.clone();
689 promise.set_waiter(Rc::new(move || {
690 if let Some(state) = weak.upgrade() {
691 loop {
692 if !state.is_pending(task) {
693 break;
694 }
695 if let Err(error) = state.pump(&manifest_for_wait) {
696 state.fail_all(error);
697 break;
698 }
699 if !state.is_pending(task) {
700 break;
701 }
702 std::thread::sleep(Duration::from_millis(1));
703 }
704 if matches!(waiting.state(), PromiseState::Pending) && state.is_expired(task) {
705 state.timeout(task);
706 }
707 }
708 }));
709 let weak = Rc::downgrade(self);
710 promise.set_cancel_hook(Rc::new(move || {
711 if let Some(state) = weak.upgrade() {
712 let _ = state.cancel(task);
713 }
714 }));
715 if let Err(error) = self.pump(manifest) {
716 self.fail_all(error.clone());
717 return Err(error);
718 }
719 Ok(Value::Promise(promise))
720 }
721
722 fn release(&self, manifest: &ExtensionManifest, handle: &Value) -> Result<(), String> {
723 validate_handles_in_value(handle, manifest)?;
724 let Value::Extension(handle_value) = handle else {
725 return Err("hta/handle-invalid: release expects an opaque handle".into());
726 };
727 let key = (
728 handle_value.provider.clone(),
729 handle_value.type_name.clone(),
730 handle_value.handle,
731 );
732 let frame = hta::encode(handle)?;
733 let mut session_ref = self.session.borrow_mut();
734 let session = session_ref
735 .as_mut()
736 .ok_or_else(|| "hta/session-closed".to_owned())?;
737 if !session.handles.remove(&key) {
738 self.trace.emit(
739 "release",
740 None,
741 None,
742 Some("error"),
743 Some("hta/handle-stale".into()),
744 );
745 return Err(format!(
746 "hta/handle-stale: {}:{}",
747 handle_value.type_name, handle_value.handle
748 ));
749 }
750 if let Err(error) = execute_release(session, &frame) {
751 session.handles.insert(key);
752 self.trace
753 .emit("release", None, None, Some("error"), Some(error.clone()));
754 return Err(error);
755 }
756 self.trace.emit("release", None, None, Some("ok"), None);
757 Ok(())
758 }
759
760 fn pump(self: &Rc<Self>, manifest: &ExtensionManifest) -> Result<(), String> {
761 self.poll_host_promises();
762 self.deliver_pending()?;
763 loop {
764 let event = self.next_event()?;
765 let Some(event) = event else {
766 self.expire_pending();
767 return Ok(());
768 };
769 self.handle_event(manifest, event)?;
770 self.poll_host_promises();
771 self.deliver_pending()?;
772 }
773 }
774
775 fn next_event(&self) -> Result<Option<Value>, String> {
776 let mut session_ref = self.session.borrow_mut();
777 let session = session_ref
778 .as_mut()
779 .ok_or_else(|| "hta/session-closed".to_owned())?;
780 let packed = call_i64(
781 &mut session.store,
782 &session.next_event,
783 &[],
784 "hta_next_event",
785 )?;
786 if packed == 0 {
787 return Ok(None);
788 }
789 if packed < 0 {
790 return Err("hta/event-pointer-invalid".into());
791 }
792 let packed = packed as u64;
793 let pointer = (packed >> 32) as usize;
794 let size = (packed & u64::from(u32::MAX)) as usize;
795 if size == 0 || size > MAX_HTA_FRAME_BYTES {
796 return Err("hta/event-size-invalid".into());
797 }
798 let mut bytes = vec![0_u8; size];
799 session
800 .memory
801 .read(&session.store, pointer, &mut bytes)
802 .map_err(|error| format!("hta/event-memory-invalid: {error}"))?;
803 call_void(
804 &mut session.store,
805 &session.deallocator,
806 &[Val::I32(pointer as i32), Val::I32(size as i32)],
807 "hta_dealloc",
808 )?;
809 hta::decode_canonical(&bytes)
810 .map(Some)
811 .map_err(|error| format!("hta/event-malformed: {error}"))
812 }
813
814 fn handle_event(
815 self: &Rc<Self>,
816 manifest: &ExtensionManifest,
817 event: Value,
818 ) -> Result<(), String> {
819 let values = match event {
820 Value::Vector(values) => values.iter().cloned().collect::<Vec<_>>(),
821 Value::List(values) => values.iter().cloned().collect::<Vec<_>>(),
822 _ => return Err("hta/event-malformed".into()),
823 };
824 let kind = number(&values, 0, "kind")?;
825 match kind {
826 0 | 1 => {
827 let task = number(&values, 1, "task")?;
828 let payload = values
829 .get(2)
830 .cloned()
831 .ok_or_else(|| "hta/event-malformed: payload".to_owned())?;
832 validate_handles_in_value(&payload, manifest)?;
833 {
834 let session_ref = self.session.borrow();
835 let session = session_ref
836 .as_ref()
837 .ok_or_else(|| "hta/session-closed".to_owned())?;
838 if kind == 1 {
839 validate_live_handles(&payload, &session.handles)?;
840 }
841 }
842 if self.is_pending(task) {
843 self.drop_task(task)?;
844 let pending = self
845 .session
846 .borrow_mut()
847 .as_mut()
848 .and_then(|session| session.pending.remove(&task));
849 let Some(pending) = pending else {
850 return Ok(());
851 };
852 if kind == 0 {
853 if let Some(session) = self.session.borrow_mut().as_mut() {
854 collect_handles(&payload, &mut session.handles);
855 }
856 }
857 self.trace.emit(
858 if kind == 0 {
859 "call-return"
860 } else {
861 "call-error"
862 },
863 Some(task),
864 Some(pending.operation.clone()),
865 Some(if kind == 0 { "ok" } else { "error" }),
866 None,
867 );
868 if kind == 0 {
869 pending.promise.resolve(payload);
870 } else {
871 pending.promise.reject_value(payload);
872 }
873 }
874 Ok(())
875 }
876 2 => self.handle_host_event(manifest, &values),
877 _ => Err(format!("hta/event-unknown: {kind}")),
878 }
879 }
880
881 fn handle_host_event(
882 self: &Rc<Self>,
883 manifest: &ExtensionManifest,
884 values: &[Value],
885 ) -> Result<(), String> {
886 if values.len() != 6 && values.len() != 8 {
887 return Err("hta/host-call-malformed".into());
888 }
889 let call = number(values, 1, "call")?;
890 let task = number(values, 2, "task")?;
891 if !self.is_pending(task) {
892 return Ok(());
893 }
894 if !self
895 .session
896 .borrow_mut()
897 .as_mut()
898 .ok_or_else(|| "hta/session-closed".to_owned())?
899 .host_calls_seen
900 .insert(call)
901 {
902 return Ok(());
903 }
904 let service_index = if values.len() == 8 { 5 } else { 3 };
905 let service = string_value(values, service_index, "service")?;
906 let method = string_value(values, service_index + 1, "method")?;
907 let arguments = match values.get(service_index + 2) {
908 Some(Value::Vector(arguments)) => arguments.iter().cloned().collect::<Vec<_>>(),
909 Some(Value::List(arguments)) => arguments.iter().cloned().collect::<Vec<_>>(),
910 _ => return Err("hta/host-call-malformed: arguments".into()),
911 };
912 validate_handles_in_value(&Value::Vector(arguments.clone().into()), manifest)?;
913 if let Some(session) = self.session.borrow().as_ref() {
914 validate_live_handles(&Value::Vector(arguments.clone().into()), &session.handles)?;
915 }
916 if !manifest.permits_host_call(&service, &method) {
917 self.queue_delivery(
918 call,
919 false,
920 host_error("hta/host-call-denied", &service, &method),
921 );
922 return Ok(());
923 }
924 if manifest
925 .host_call_capabilities(&service, &method)
926 .iter()
927 .any(|capability| !hta_capabilities().contains(capability))
928 {
929 self.queue_delivery(
930 call,
931 false,
932 host_error("hta/capability-denied", &service, &method),
933 );
934 return Ok(());
935 }
936 let Some(handler) = self.host_handler.clone() else {
937 self.queue_delivery(
938 call,
939 false,
940 host_error("host/unavailable", &service, &method),
941 );
942 return Ok(());
943 };
944 match handler(service.clone(), method.clone(), arguments) {
945 Ok(Value::Promise(promise)) => {
946 self.session
947 .borrow_mut()
948 .as_mut()
949 .ok_or_else(|| "hta/session-closed".to_owned())?
950 .host_promises
951 .insert(call, promise.clone());
952 let weak = Rc::downgrade(self);
953 promise.on_settle(Rc::new(move |state| {
954 if let Some(state_owner) = weak.upgrade() {
955 match state {
956 PromiseState::Fulfilled(value) => {
957 state_owner.queue_delivery(call, true, value)
958 }
959 PromiseState::Rejected(error) => state_owner.queue_delivery(
960 call,
961 false,
962 host_failure("hta/host-call-failed", &error.message()),
963 ),
964 PromiseState::Pending => {}
965 }
966 }
967 }));
968 }
969 Ok(value) => self.queue_delivery(call, true, value),
970 Err(error) => {
971 self.queue_delivery(call, false, host_failure("hta/host-call-failed", &error))
972 }
973 }
974 Ok(())
975 }
976
977 fn poll_host_promises(&self) {
978 let promises = self
979 .session
980 .borrow()
981 .as_ref()
982 .map(|session| session.host_promises.values().cloned().collect::<Vec<_>>())
983 .unwrap_or_default();
984 for promise in promises {
985 let _ = promise.state();
986 }
987 }
988
989 fn queue_delivery(&self, call: u64, fulfilled: bool, value: Value) {
990 if let Some(session) = self.session.borrow_mut().as_mut() {
991 session.deliveries.push_back((call, fulfilled, value));
992 }
993 }
994
995 fn deliver_pending(&self) -> Result<(), String> {
996 loop {
997 let delivery = self
998 .session
999 .borrow_mut()
1000 .as_mut()
1001 .and_then(|session| session.deliveries.pop_front());
1002 let Some((call, fulfilled, value)) = delivery else {
1003 return Ok(());
1004 };
1005 let frame = hta::encode(&Value::Vector(
1006 vec![
1007 Value::Number(call as i64),
1008 Value::Number(if fulfilled { 0 } else { 1 }),
1009 value.clone(),
1010 ]
1011 .into(),
1012 ))?;
1013 let mut session_ref = self.session.borrow_mut();
1014 let session = session_ref
1015 .as_mut()
1016 .ok_or_else(|| "hta/session-closed".to_owned())?;
1017 execute_deliver(session, &frame)?;
1018 if fulfilled {
1019 collect_handles(&value, &mut session.handles);
1020 }
1021 session.host_promises.remove(&call);
1022 }
1023 }
1024
1025 fn drop_task(&self, task: u64) -> Result<(), String> {
1026 let mut session_ref = self.session.borrow_mut();
1027 let session = session_ref
1028 .as_mut()
1029 .ok_or_else(|| "hta/session-closed".to_owned())?;
1030 let status = call_i32(
1031 &mut session.store,
1032 &session.drop_task,
1033 &[Val::I64(task as i64)],
1034 "hta_drop_task",
1035 )?;
1036 if status != 0 {
1037 return Err(format!("hta/drop-task-failed: {status}"));
1038 }
1039 Ok(())
1040 }
1041
1042 fn cancel(&self, task: u64) -> Result<(), String> {
1043 let pending = self
1044 .session
1045 .borrow_mut()
1046 .as_mut()
1047 .ok_or_else(|| "hta/session-closed".to_owned())?
1048 .pending
1049 .remove(&task);
1050 let Some(pending) = pending else {
1051 return Ok(());
1052 };
1053 if let Err(error) = self.cancel_task(task) {
1054 let _ = self.drop_task(task);
1055 self.trace.emit(
1056 "cancel",
1057 Some(task),
1058 Some(pending.operation),
1059 Some("error"),
1060 Some(error.clone()),
1061 );
1062 return Err(error);
1063 }
1064 self.trace.emit(
1065 "cancel",
1066 Some(task),
1067 Some(pending.operation),
1068 Some("ok"),
1069 None,
1070 );
1071 Ok(())
1072 }
1073
1074 fn cancel_task(&self, task: u64) -> Result<(), String> {
1075 let mut session_ref = self.session.borrow_mut();
1076 let session = session_ref
1077 .as_mut()
1078 .ok_or_else(|| "hta/session-closed".to_owned())?;
1079 let status = call_i32(
1080 &mut session.store,
1081 &session.cancel,
1082 &[Val::I64(task as i64)],
1083 "hta_cancel",
1084 )?;
1085 if status != 0 {
1086 return Err(format!("hta/cancel-failed: {status}"));
1087 }
1088 let drop_status = call_i32(
1089 &mut session.store,
1090 &session.drop_task,
1091 &[Val::I64(task as i64)],
1092 "hta_drop_task",
1093 )?;
1094 if drop_status != 0 {
1095 return Err(format!("hta/drop-task-failed: {drop_status}"));
1096 }
1097 Ok(())
1098 }
1099
1100 fn is_pending(&self, task: u64) -> bool {
1101 self.session
1102 .borrow()
1103 .as_ref()
1104 .is_some_and(|session| session.pending.contains_key(&task))
1105 }
1106
1107 fn is_expired(&self, task: u64) -> bool {
1108 self.session
1109 .borrow()
1110 .as_ref()
1111 .and_then(|session| session.pending.get(&task))
1112 .and_then(|pending| pending.deadline)
1113 .is_some_and(|deadline| deadline <= Instant::now())
1114 }
1115
1116 fn expire_pending(&self) {
1117 let expired = self
1118 .session
1119 .borrow()
1120 .as_ref()
1121 .map(|session| {
1122 session
1123 .pending
1124 .iter()
1125 .filter_map(|(task, pending)| {
1126 pending
1127 .deadline
1128 .filter(|deadline| *deadline <= Instant::now())
1129 .map(|_| *task)
1130 })
1131 .collect::<Vec<_>>()
1132 })
1133 .unwrap_or_default();
1134 for task in expired {
1135 self.timeout(task);
1136 }
1137 }
1138
1139 fn timeout(&self, task: u64) {
1140 if self.is_pending(task) {
1141 if self.cancel_task(task).is_err() {
1142 let _ = self.drop_task(task);
1143 }
1144 let pending = self
1145 .session
1146 .borrow_mut()
1147 .as_mut()
1148 .and_then(|session| session.pending.remove(&task));
1149 if let Some(pending) = pending {
1150 self.trace.emit(
1151 "call-error",
1152 Some(task),
1153 Some(pending.operation),
1154 Some("error"),
1155 Some("hta/timeout".into()),
1156 );
1157 pending.promise.notify_cancel();
1158 pending.promise.reject("hta/timeout");
1159 }
1160 }
1161 }
1162
1163 fn fail_all(&self, error: String) {
1164 let pending = self
1165 .session
1166 .borrow_mut()
1167 .as_mut()
1168 .map(|session| {
1169 session
1170 .pending
1171 .drain()
1172 .map(|(task, pending)| (task, pending.promise))
1173 .collect::<Vec<_>>()
1174 })
1175 .unwrap_or_default();
1176 for (task, promise) in pending {
1177 if let Some(session) = self.session.borrow_mut().as_mut() {
1178 let _ = cancel_task_on_session(session, task);
1179 let _ = drop_task_on_session(session, task);
1180 }
1181 promise.reject(error.clone());
1182 }
1183 if let Some(session) = self.session.borrow_mut().as_mut() {
1184 session.host_promises.clear();
1185 session.deliveries.clear();
1186 }
1187 }
1188
1189 fn shutdown(&self, _manifest: &ExtensionManifest) {
1190 let Some(mut session) = self.session.borrow_mut().take() else {
1191 self.trace.emit_shutdown(Some("ok"), None);
1192 return;
1193 };
1194 let pending = session
1195 .pending
1196 .drain()
1197 .map(|(task, pending)| (task, pending.operation, pending.promise))
1198 .collect::<Vec<_>>();
1199 for (task, operation, promise) in pending {
1200 let _ = cancel_task_on_session(&mut session, task);
1201 let _ = drop_task_on_session(&mut session, task);
1202 self.trace.emit(
1203 "call-error",
1204 Some(task),
1205 Some(operation),
1206 Some("error"),
1207 Some("hta/session-closed".into()),
1208 );
1209 promise.reject("hta/session-closed");
1210 }
1211 session.host_promises.clear();
1212 session.deliveries.clear();
1213 self.trace.emit_shutdown(Some("ok"), None);
1214 }
1215}
1216
1217fn validate_handles_in_value(value: &Value, manifest: &ExtensionManifest) -> Result<(), String> {
1218 match value {
1219 Value::Extension(handle) => {
1220 if manifest.handle_tags.is_empty() {
1221 return Ok(());
1222 }
1223 let Some(owner) = manifest.handle_tags.get(&handle.type_name) else {
1224 return Err(format!("hta/handle-type-denied: {}", handle.type_name));
1225 };
1226 if handle.provider != manifest.namespace
1227 && manifest.identity.as_deref() != Some(handle.provider.as_str())
1228 && handle.provider != *owner
1229 {
1230 return Err(format!(
1231 "hta/handle-owner-mismatch: {}:{}",
1232 handle.provider, handle.handle
1233 ));
1234 }
1235 }
1236 Value::Tagged(value) => validate_handles_in_value(value.form(), manifest)?,
1237 Value::Vector(values) => validate_handles_iter(values.iter(), manifest)?,
1238 Value::List(values) => validate_handles_iter(values.iter(), manifest)?,
1239 Value::Tuple(values) => validate_handles_iter(values.iter(), manifest)?,
1240 Value::MapEntry(entry) => {
1241 validate_handles_in_value(entry.key(), manifest)?;
1242 validate_handles_in_value(entry.value(), manifest)?;
1243 }
1244 Value::Map(values) => validate_handles_map(values.iter(), manifest)?,
1245 Value::SortedMap(values) => validate_handles_map(values.iter(), manifest)?,
1246 Value::OrderedMap(values) => {
1247 for (key, value) in values.iter() {
1248 validate_handles_in_value(key, manifest)?;
1249 validate_handles_in_value(value, manifest)?;
1250 }
1251 }
1252 Value::PriorityMap(values) => {
1253 for (key, value) in values.iter() {
1254 validate_handles_in_value(&key, manifest)?;
1255 validate_handles_in_value(&value, manifest)?;
1256 }
1257 }
1258 Value::Set(values) => validate_handles_iter(values.iter(), manifest)?,
1259 Value::OrderedSet(values) => validate_handles_iter(values.iter(), manifest)?,
1260 Value::SortedSet(values) => validate_handles_iter(values.iter(), manifest)?,
1261 Value::Struct(value) => {
1262 for value in value.ordered_values() {
1263 validate_handles_in_value(value, manifest)?;
1264 }
1265 }
1266 _ => {}
1267 }
1268 Ok(())
1269}
1270
1271fn validate_handles_iter<'a>(
1272 values: impl Iterator<Item = &'a Value>,
1273 manifest: &ExtensionManifest,
1274) -> Result<(), String> {
1275 for value in values {
1276 validate_handles_in_value(value, manifest)?;
1277 }
1278 Ok(())
1279}
1280
1281fn validate_handles_map<'a>(
1282 values: impl Iterator<Item = (&'a Value, &'a Value)>,
1283 manifest: &ExtensionManifest,
1284) -> Result<(), String> {
1285 for (key, value) in values {
1286 validate_handles_in_value(key, manifest)?;
1287 validate_handles_in_value(value, manifest)?;
1288 }
1289 Ok(())
1290}
1291
1292fn validate_live_handles(
1293 value: &Value,
1294 handles: &HashSet<(String, String, u64)>,
1295) -> Result<(), String> {
1296 match value {
1297 Value::Extension(handle)
1298 if !handles.contains(&(
1299 handle.provider.clone(),
1300 handle.type_name.clone(),
1301 handle.handle,
1302 )) =>
1303 {
1304 return Err(format!(
1305 "hta/handle-stale: {}:{}",
1306 handle.type_name, handle.handle
1307 ));
1308 }
1309 Value::Vector(values) => validate_live_iter(values.iter(), handles)?,
1310 Value::List(values) => validate_live_iter(values.iter(), handles)?,
1311 Value::Tuple(values) => validate_live_iter(values.iter(), handles)?,
1312 Value::MapEntry(entry) => {
1313 validate_live_handles(entry.key(), handles)?;
1314 validate_live_handles(entry.value(), handles)?;
1315 }
1316 Value::Map(values) => validate_live_map(values.iter(), handles)?,
1317 Value::SortedMap(values) => validate_live_map(values.iter(), handles)?,
1318 Value::OrderedMap(values) => {
1319 for (key, value) in values.iter() {
1320 validate_live_handles(key, handles)?;
1321 validate_live_handles(value, handles)?;
1322 }
1323 }
1324 Value::PriorityMap(values) => {
1325 for (key, value) in values.iter() {
1326 validate_live_handles(&key, handles)?;
1327 validate_live_handles(&value, handles)?;
1328 }
1329 }
1330 Value::Set(values) => validate_live_iter(values.iter(), handles)?,
1331 Value::OrderedSet(values) => validate_live_iter(values.iter(), handles)?,
1332 Value::SortedSet(values) => validate_live_iter(values.iter(), handles)?,
1333 Value::Tagged(value) => validate_live_handles(value.form(), handles)?,
1334 Value::Struct(value) => {
1335 for value in value.ordered_values() {
1336 validate_live_handles(value, handles)?;
1337 }
1338 }
1339 _ => {}
1340 }
1341 Ok(())
1342}
1343
1344fn validate_live_iter<'a>(
1345 values: impl Iterator<Item = &'a Value>,
1346 handles: &HashSet<(String, String, u64)>,
1347) -> Result<(), String> {
1348 for value in values {
1349 validate_live_handles(value, handles)?;
1350 }
1351 Ok(())
1352}
1353
1354fn validate_live_map<'a>(
1355 values: impl Iterator<Item = (&'a Value, &'a Value)>,
1356 handles: &HashSet<(String, String, u64)>,
1357) -> Result<(), String> {
1358 for (key, value) in values {
1359 validate_live_handles(key, handles)?;
1360 validate_live_handles(value, handles)?;
1361 }
1362 Ok(())
1363}
1364
1365fn collect_handles(value: &Value, handles: &mut HashSet<(String, String, u64)>) {
1366 match value {
1367 Value::Extension(handle) => {
1368 handles.insert((
1369 handle.provider.clone(),
1370 handle.type_name.clone(),
1371 handle.handle,
1372 ));
1373 }
1374 Value::Vector(values) => collect_iter(values.iter(), handles),
1375 Value::List(values) => collect_iter(values.iter(), handles),
1376 Value::Tuple(values) => collect_iter(values.iter(), handles),
1377 Value::MapEntry(entry) => {
1378 collect_handles(entry.key(), handles);
1379 collect_handles(entry.value(), handles);
1380 }
1381 Value::Map(values) => collect_map(values.iter(), handles),
1382 Value::SortedMap(values) => collect_map(values.iter(), handles),
1383 Value::OrderedMap(values) => {
1384 for (key, value) in values.iter() {
1385 collect_handles(key, handles);
1386 collect_handles(value, handles);
1387 }
1388 }
1389 Value::PriorityMap(values) => {
1390 for (key, value) in values.iter() {
1391 collect_handles(&key, handles);
1392 collect_handles(&value, handles);
1393 }
1394 }
1395 Value::Set(values) => collect_iter(values.iter(), handles),
1396 Value::OrderedSet(values) => collect_iter(values.iter(), handles),
1397 Value::SortedSet(values) => collect_iter(values.iter(), handles),
1398 Value::Tagged(value) => collect_handles(value.form(), handles),
1399 Value::Struct(value) => {
1400 for value in value.ordered_values() {
1401 collect_handles(value, handles);
1402 }
1403 }
1404 _ => {}
1405 }
1406}
1407
1408fn collect_iter<'a>(
1409 values: impl Iterator<Item = &'a Value>,
1410 handles: &mut HashSet<(String, String, u64)>,
1411) {
1412 for value in values {
1413 collect_handles(value, handles);
1414 }
1415}
1416
1417fn collect_map<'a>(
1418 values: impl Iterator<Item = (&'a Value, &'a Value)>,
1419 handles: &mut HashSet<(String, String, u64)>,
1420) {
1421 for (key, value) in values {
1422 collect_handles(key, handles);
1423 collect_handles(value, handles);
1424 }
1425}
1426
1427fn compile_hta_module(bytes: &[u8], allow_library: bool) -> Result<(Engine, Module), String> {
1428 let mut config = Config::new();
1429 config.consume_fuel(true);
1430 let engine =
1431 Engine::new(&config).map_err(|error| format!("extension/engine-unavailable: {error}"))?;
1432 let module = Module::new(&engine, bytes)
1433 .map_err(|error| format!("extension/module-invalid: {error}"))?;
1434 for import in module.imports() {
1435 let supported_env = import.module() == "env"
1436 && matches!(
1437 import.name(),
1438 "hara_random_fill" | "hara_time_ms" | "hara_time_ns"
1439 );
1440 let supported_library = allow_library && import.module() == "hara/library";
1441 if !supported_env && !supported_library {
1442 return Err(format!(
1443 "extension/module-invalid: unsupported import {}::{}",
1444 import.module(),
1445 import.name()
1446 ));
1447 }
1448 }
1449 Ok((engine, module))
1450}
1451
1452fn require_export(
1453 instance: &Instance,
1454 store: &mut Store<StoreLimits>,
1455 name: &str,
1456) -> Result<Func, String> {
1457 instance
1458 .get_func(&mut *store, name)
1459 .ok_or_else(|| format!("extension/malformed: module has no export {name}"))
1460}
1461
1462fn expect_signature(
1463 store: &mut Store<StoreLimits>,
1464 function: &Func,
1465 parameters: &[ValType],
1466 results: &[ValType],
1467 name: &str,
1468) -> Result<(), String> {
1469 let ty = function.ty(&mut *store);
1470 let actual_parameters = ty.params().collect::<Vec<_>>();
1471 let actual_results = ty.results().collect::<Vec<_>>();
1472 if actual_parameters != parameters || actual_results != results {
1473 return Err(format!(
1474 "extension/abi-type-unsupported: {name} has an invalid signature"
1475 ));
1476 }
1477 Ok(())
1478}
1479
1480fn call_i32(
1481 store: &mut Store<StoreLimits>,
1482 function: &Func,
1483 arguments: &[Val],
1484 name: &str,
1485) -> Result<i32, String> {
1486 store
1487 .set_fuel(10_000_000)
1488 .map_err(|error| format!("extension/execution-limit: {error}"))?;
1489 let mut results = [Val::I32(0)];
1490 function
1491 .call(store, arguments, &mut results)
1492 .map_err(|error| format!("extension/{name}-failed: {error}"))?;
1493 match results[0] {
1494 Val::I32(value) => Ok(value),
1495 _ => Err(format!("extension/abi-type-unsupported: {name}")),
1496 }
1497}
1498
1499fn call_i64(
1500 store: &mut Store<StoreLimits>,
1501 function: &Func,
1502 arguments: &[Val],
1503 name: &str,
1504) -> Result<i64, String> {
1505 store
1506 .set_fuel(10_000_000)
1507 .map_err(|error| format!("extension/execution-limit: {error}"))?;
1508 let mut results = [Val::I64(0)];
1509 function
1510 .call(store, arguments, &mut results)
1511 .map_err(|error| format!("extension/{name}-failed: {error}"))?;
1512 match results[0] {
1513 Val::I64(value) => Ok(value),
1514 _ => Err(format!("extension/abi-type-unsupported: {name}")),
1515 }
1516}
1517
1518fn call_void(
1519 store: &mut Store<StoreLimits>,
1520 function: &Func,
1521 arguments: &[Val],
1522 name: &str,
1523) -> Result<(), String> {
1524 store
1525 .set_fuel(10_000_000)
1526 .map_err(|error| format!("extension/execution-limit: {error}"))?;
1527 function
1528 .call(store, arguments, &mut [])
1529 .map_err(|error| format!("extension/{name}-failed: {error}"))
1530}
1531
1532fn execute_start(session: &mut HtaSession, frame: &[u8]) -> Result<i64, String> {
1533 let pointer = call_i32(
1534 &mut session.store,
1535 &session.allocator,
1536 &[Val::I32(frame.len() as i32)],
1537 "hta_alloc",
1538 )?;
1539 if pointer < 0 {
1540 return Err("hta/memory-unavailable".into());
1541 }
1542 session
1543 .memory
1544 .write(&mut session.store, pointer as usize, frame)
1545 .map_err(|error| format!("hta/memory-write-failed: {error}"))?;
1546 let result = call_i64(
1547 &mut session.store,
1548 &session.start,
1549 &[Val::I32(pointer), Val::I32(frame.len() as i32)],
1550 "hta_start",
1551 );
1552 call_void(
1553 &mut session.store,
1554 &session.deallocator,
1555 &[Val::I32(pointer), Val::I32(frame.len() as i32)],
1556 "hta_dealloc",
1557 )?;
1558 result
1559}
1560
1561fn execute_deliver(session: &mut HtaSession, frame: &[u8]) -> Result<(), String> {
1562 let pointer = call_i32(
1563 &mut session.store,
1564 &session.allocator,
1565 &[Val::I32(frame.len() as i32)],
1566 "hta_alloc",
1567 )?;
1568 if pointer < 0 {
1569 return Err("hta/memory-unavailable".into());
1570 }
1571 session
1572 .memory
1573 .write(&mut session.store, pointer as usize, frame)
1574 .map_err(|error| format!("hta/memory-write-failed: {error}"))?;
1575 let status = call_i32(
1576 &mut session.store,
1577 &session.deliver,
1578 &[Val::I32(pointer), Val::I32(frame.len() as i32)],
1579 "hta_deliver",
1580 );
1581 call_void(
1582 &mut session.store,
1583 &session.deallocator,
1584 &[Val::I32(pointer), Val::I32(frame.len() as i32)],
1585 "hta_dealloc",
1586 )?;
1587 let status = status?;
1588 if status != 0 {
1589 return Err(format!("hta/deliver-failed: {status}"));
1590 }
1591 Ok(())
1592}
1593
1594fn execute_release(session: &mut HtaSession, frame: &[u8]) -> Result<(), String> {
1595 let pointer = call_i32(
1596 &mut session.store,
1597 &session.allocator,
1598 &[Val::I32(frame.len() as i32)],
1599 "hta_alloc",
1600 )?;
1601 if pointer < 0 {
1602 return Err("hta/memory-unavailable".into());
1603 }
1604 session
1605 .memory
1606 .write(&mut session.store, pointer as usize, frame)
1607 .map_err(|error| format!("hta/memory-write-failed: {error}"))?;
1608 let status = call_i32(
1609 &mut session.store,
1610 &session.release,
1611 &[Val::I32(pointer), Val::I32(frame.len() as i32)],
1612 "hta_release",
1613 );
1614 call_void(
1615 &mut session.store,
1616 &session.deallocator,
1617 &[Val::I32(pointer), Val::I32(frame.len() as i32)],
1618 "hta_dealloc",
1619 )?;
1620 let status = status?;
1621 if status != 0 {
1622 return Err(format!("hta/handle-release-failed: {status}"));
1623 }
1624 Ok(())
1625}
1626
1627fn drop_task_on_session(session: &mut HtaSession, task: u64) -> Result<(), String> {
1628 let status = call_i32(
1629 &mut session.store,
1630 &session.drop_task,
1631 &[Val::I64(task as i64)],
1632 "hta_drop_task",
1633 )?;
1634 if status != 0 {
1635 return Err(format!("hta/drop-task-failed: {status}"));
1636 }
1637 Ok(())
1638}
1639
1640fn cancel_task_on_session(session: &mut HtaSession, task: u64) -> Result<(), String> {
1641 let status = call_i32(
1642 &mut session.store,
1643 &session.cancel,
1644 &[Val::I64(task as i64)],
1645 "hta_cancel",
1646 )?;
1647 if status != 0 {
1648 return Err(format!("hta/cancel-failed: {status}"));
1649 }
1650 Ok(())
1651}
1652
1653fn hta_timeout() -> Option<Duration> {
1654 match std::env::var("HARA_HTA_TIMEOUT_MS") {
1655 Ok(value) => match value.parse::<u64>() {
1656 Ok(0) => None,
1657 Ok(milliseconds) => Some(Duration::from_millis(milliseconds)),
1658 Err(_) => Some(DEFAULT_HTA_TIMEOUT),
1659 },
1660 Err(_) => Some(DEFAULT_HTA_TIMEOUT),
1661 }
1662}
1663
1664fn number(values: &[Value], index: usize, field: &str) -> Result<u64, String> {
1665 match values.get(index) {
1666 Some(Value::Number(value)) if *value >= 0 => Ok(*value as u64),
1667 _ => Err(format!("hta/event-malformed: {field}")),
1668 }
1669}
1670
1671fn string_value(values: &[Value], index: usize, field: &str) -> Result<String, String> {
1672 match values.get(index) {
1673 Some(Value::String(value)) => Ok(value.clone()),
1674 _ => Err(format!("hta/event-malformed: {field}")),
1675 }
1676}
1677
1678fn host_error(code: &str, service: &str, method: &str) -> Value {
1679 host_failure(code, &format!("{service}/{method}"))
1680}
1681
1682fn host_failure(code: &str, message: &str) -> Value {
1683 Value::Map(
1684 [
1685 (Value::Keyword("code".into()), Value::Keyword(code.into())),
1686 (
1687 Value::Keyword("message".into()),
1688 Value::String(message.into()),
1689 ),
1690 (
1691 Value::Keyword("origin".into()),
1692 Value::Keyword("host".into()),
1693 ),
1694 (Value::Keyword("retryable".into()), Value::Bool(false)),
1695 ]
1696 .into_iter()
1697 .collect(),
1698 )
1699}
1700
1701fn argument(export: &str, wire_type: &str, value: &Value) -> Result<Val, String> {
1702 fn finite_f32(value: f64) -> Result<f32, String> {
1703 let value = value as f32;
1704 if value.is_finite() {
1705 Ok(value)
1706 } else {
1707 Err("non-finite number".into())
1708 }
1709 }
1710 let type_error = || format!("extension/type-error: {export} expects {wire_type}");
1711 match (wire_type, value) {
1712 ("i32", Value::Number(value)) => i32::try_from(*value)
1713 .map(Val::I32)
1714 .map_err(|_| type_error()),
1715 ("i64", Value::Number(value)) => Ok(Val::I64(*value)),
1716 ("f32", Value::Float(value)) => Ok(Val::F32(finite_f32(*value)?.to_bits())),
1717 ("f32", Value::Number(value)) => Ok(Val::F32(finite_f32(*value as f64)?.to_bits())),
1718 ("f64", Value::Float(value)) => {
1719 Ok(Val::F64(crate::numeric::finite_float(*value)?.to_bits()))
1720 }
1721 ("f64", Value::Number(value)) => Ok(Val::F64((*value as f64).to_bits())),
1722 ("boolean", Value::Bool(value)) => Ok(Val::I32(i32::from(*value))),
1723 _ => Err(type_error()),
1724 }
1725}
1726
1727fn default_result(wire_type: &str) -> Result<Val, String> {
1728 match wire_type {
1729 "i32" | "boolean" => Ok(Val::I32(0)),
1730 "i64" => Ok(Val::I64(0)),
1731 "f32" => Ok(Val::F32(0)),
1732 "f64" => Ok(Val::F64(0)),
1733 _ => Err(format!("extension/abi-type-unsupported: {wire_type}")),
1734 }
1735}
1736
1737fn result(export: &str, wire_type: &str, value: Option<Val>) -> Result<Value, String> {
1738 match (wire_type, value) {
1739 ("void", None) => Ok(Value::Nil),
1740 ("i32", Some(Val::I32(value))) => Ok(Value::Number(i64::from(value))),
1741 ("i64", Some(Val::I64(value))) => Ok(Value::Number(value)),
1742 ("f32", Some(Val::F32(value))) => Ok(Value::Float(crate::numeric::finite_float(
1743 f32::from_bits(value) as f64,
1744 )?)),
1745 ("f64", Some(Val::F64(value))) => Ok(Value::Float(crate::numeric::finite_float(
1746 f64::from_bits(value),
1747 )?)),
1748 ("boolean", Some(Val::I32(value))) => Ok(Value::Bool(value != 0)),
1749 _ => Err(format!(
1750 "extension/abi-type-unsupported: {export} -> {wire_type}"
1751 )),
1752 }
1753}
1754
1755#[cfg(test)]
1756mod tests {
1757 use crate::extension::{ExtensionManifest, Value, WasmExtension};
1758
1759 use super::{HtaProviderTrace, WasmtimeExtensionProvider, HTA_PROVIDER_EVENT_SCHEMA};
1760
1761 const ADD: &[u8] = b"\0asm\x01\0\0\0\x01\x07\x01\x60\x02\x7e\x7e\x01\x7e\x03\x02\x01\0\x07\x07\x01\x03add\0\0\x0a\x09\x01\x07\0\x20\0\x20\x01\x7c\x0b";
1762 const ALIASED_MANIFEST: &str = r#"
1763 {:namespace "math.scalar"
1764 :version "0.1.0"
1765 :provider :wasm
1766 :module "math.wasm"
1767 :abi :core.v1
1768 :exports {"sum" {:wasm/export "add"
1769 :args [:i64 :i64]
1770 :returns :i64}}
1771 :capabilities []}"#;
1772
1773 #[test]
1774 fn invokes_a_raw_wasm_export_through_a_public_hara_name() {
1775 let manifest = ExtensionManifest::parse(ALIASED_MANIFEST, "fixture").unwrap();
1776 let provider = WasmtimeExtensionProvider::compile(ADD).unwrap();
1777 let mut extension = WasmExtension::new(manifest, provider).unwrap();
1778 let bindings = extension.require().unwrap();
1779 assert_eq!(bindings[0].name, "sum");
1780 assert_eq!(
1781 bindings[0]
1782 .invoke(&[Value::Number(19), Value::Number(23)])
1783 .unwrap(),
1784 Value::Number(42)
1785 );
1786 }
1787
1788 #[test]
1789 fn hta_lifecycle_trace_is_stable_and_shutdown_is_idempotent() {
1790 let trace = HtaProviderTrace::new();
1791 trace.emit("start", None, None, Some("ok"), None);
1792 trace.emit("call-enter", Some(7), Some("demo/echo".into()), None, None);
1793 trace.emit_shutdown(Some("ok"), None);
1794 trace.emit_shutdown(Some("error"), Some("late".into()));
1795
1796 let events = trace.drain();
1797 assert_eq!(events.len(), 3);
1798 assert_eq!(events[0].schema, HTA_PROVIDER_EVENT_SCHEMA);
1799 assert_eq!(events[0].origin, "wasmtime");
1800 assert_eq!(events[1].sequence, 2);
1801 assert_eq!(events[1].request, Some(7));
1802 assert_eq!(events[1].operation.as_deref(), Some("demo/echo"));
1803 assert_eq!(events[2].event, "shutdown");
1804 assert_eq!(events[2].sequence, 3);
1805 }
1806}