scripting/

executor.rs

use nomiscript::SymbolTable;
use tracing::debug;
use wasmtime::{Engine, Linker, Module, Store};

use crate::error::HookError;
use crate::format::{BASE_OFFSET, GlobalHeader, OUTPUT_HEADER_SIZE, OutputHeader};
use crate::host::{WasmHost, define_host_functions};
use crate::parser::{OutputParser, ParsedEntity};
use crate::runtime::{EngineOpts, build_engine};

const DEFAULT_OUTPUT_SIZE: u32 = 64 * 1024;
const WASM_PAGE_SIZE: u32 = 65536;
/// Wall-clock budget for one script execution, in [`EPOCH_TICK_INTERVAL`] ticks.
///
/// This is a real-time bound, not a fuel bound, so it is sensitive to how fast
/// the host happens to be. Coverage-instrumented builds run several times
/// slower, which is why [`ScriptExecutor::with_epoch_deadline_ticks`] exists:
/// tests that only care that a script *completes* raise the budget rather than
/// racing the clock.
const DEFAULT_EPOCH_DEADLINE_TICKS: u64 = 5;
const EPOCH_TICK_INTERVAL: std::time::Duration = std::time::Duration::from_secs(1);

/// Drives the engine's epoch clock for the lifetime of one execution.
///
/// Stopping matters: the epoch counter is shared by every execution on the same
/// engine, so a ticker that outlives its call keeps consuming the budget of
/// whatever runs next. A `JoinHandle` does not stop a thread when dropped, it
/// detaches it — so the previous `drop(ticker)` left it ticking for up to
/// `EPOCH_DEADLINE_TICKS` seconds, and every `?` between the spawn and the call
/// skipped even that. Executors are reused (the account-script batch path runs
/// one across many transactions), which is exactly where a stray bump lands.
///
/// The stop channel doubles as the sleep: `recv_timeout` returns `Timeout` for a
/// real tick and `Disconnected` the moment the guard drops, so teardown is
/// prompt rather than waiting out the current second.
struct EpochTicker {
    stop: Option<std::sync::mpsc::Sender<()>>,
    handle: Option<std::thread::JoinHandle<()>>,
}

impl EpochTicker {
    fn spawn(engine: Engine, ticks: u64) -> Self {
        let (stop, rx) = std::sync::mpsc::channel::<()>();
        let handle = std::thread::spawn(move || {
            for _ in 0..ticks {
                match rx.recv_timeout(EPOCH_TICK_INTERVAL) {
                    Err(std::sync::mpsc::RecvTimeoutError::Timeout) => engine.increment_epoch(),
                    _ => return,
                }
            }
        });
        Self {
            stop: Some(stop),
            handle: Some(handle),
        }
    }
}

impl Drop for EpochTicker {
    fn drop(&mut self) {
        self.stop.take();
        if let Some(handle) = self.handle.take() {
            let _ = handle.join();
        }
    }
}

pub struct ScriptExecutor {
    host: WasmHost,
    epoch_deadline_ticks: u64,
}

impl ScriptExecutor {
    /// Builds an executor on the baseline engine.
    ///
    /// Fallible because engine construction is: `wasmtime` validates the
    /// configuration and rejects unsupported combinations. There is no
    /// infallible `new`/`Default` for the same reason — a constructor that
    /// cannot report that failure can only paper over it.
    pub fn try_new() -> Result<Self, HookError> {
        Ok(Self {
            host: WasmHost::new(build_engine(EngineOpts::baseline())?, SymbolTable::new()),
            epoch_deadline_ticks: DEFAULT_EPOCH_DEADLINE_TICKS,
        })
    }

    #[must_use]
    pub fn with_engine(engine: Engine) -> Self {
        Self {
            host: WasmHost::new(engine, SymbolTable::new()),
            epoch_deadline_ticks: DEFAULT_EPOCH_DEADLINE_TICKS,
        }
    }

    /// Overrides the per-execution wall-clock budget.
    ///
    /// Production uses [`DEFAULT_EPOCH_DEADLINE_TICKS`]; a caller that runs
    /// under coverage instrumentation, a debugger, or a loaded CI node raises
    /// it so a slow-but-correct script is not cut off mid-execution.
    #[must_use]
    pub fn with_epoch_deadline_ticks(mut self, ticks: u64) -> Self {
        self.epoch_deadline_ticks = ticks;
        self
    }

    fn get_or_compile_module(&self, bytecode: &[u8]) -> Result<Module, HookError> {
        self.host
            .module_cache()
            .get_or_compile(self.host.engine(), bytecode)
            .map_err(HookError::from)
    }

    pub fn execute(
        &self,
        bytecode: &[u8],
        input: &[u8],
        output_size: Option<u32>,
    ) -> Result<Vec<ParsedEntity>, HookError> {
        debug!(bytecode_size = bytecode.len(), "script execution start");
        let output_size = output_size.unwrap_or(DEFAULT_OUTPUT_SIZE);
        let module = self.get_or_compile_module(bytecode)?;
        debug!("module compiled");

        let header = GlobalHeader::from_bytes(input)
            .ok_or_else(|| HookError::Parse("Invalid input header".to_string()))?;

        let input_offset = BASE_OFFSET;
        let output_offset = input_offset + input.len() as u32;
        let strings_offset = header.strings_pool_offset;

        let exec_state = self
            .host
            .execution_state(input_offset, output_offset, strings_offset);
        let mut store = Store::new(self.host.engine(), exec_state);
        store.set_epoch_deadline(self.epoch_deadline_ticks);

        let _ticker = EpochTicker::spawn(self.host.engine().clone(), self.epoch_deadline_ticks);

        let mut linker = Linker::new(self.host.engine());
        define_host_functions(&mut linker)?;

        let instance = linker.instantiate(&mut store, &module)?;

        let memory = instance
            .get_memory(&mut store, "memory")
            .ok_or(HookError::WASMMem)?;

        store.data_mut().memory = Some(memory);

        let total_size = input.len() + output_size as usize;
        let required_pages = (BASE_OFFSET as usize + total_size).div_ceil(WASM_PAGE_SIZE as usize);
        let current_pages = memory.size(&store) as usize;

        if required_pages > current_pages {
            memory.grow(&mut store, (required_pages - current_pages) as u64)?;
        }

        let mem_data = memory.data_mut(&mut store);
        let input_start = BASE_OFFSET as usize;
        mem_data[input_start..input_start + input.len()].copy_from_slice(input);

        let output_start = output_offset as usize;
        let input_entity_count = header.input_entity_count;
        let output_header = OutputHeader::new(input_entity_count);
        mem_data[output_start..output_start + OUTPUT_HEADER_SIZE]
            .copy_from_slice(&output_header.to_bytes());

        let should_apply = instance
            .get_typed_func::<(), i32>(&mut store, "should_apply")
            .map_err(|e| HookError::Script(format!("Missing should_apply export: {e}")))?;

        let result = should_apply.call(&mut store, ())?;
        debug!(should_apply = result, "should_apply result");
        if result == 0 {
            return Ok(Vec::new());
        }

        let process = instance
            .get_typed_func::<(), ()>(&mut store, "process")
            .map_err(|e| HookError::Script(format!("Missing process export: {e}")))?;

        debug!("calling process");
        process.call(&mut store, ())?;

        let mem_data = memory.data(&store);
        let output_data = &mem_data[output_start..output_start + output_size as usize];

        let output_header = OutputHeader::from_bytes(output_data)
            .ok_or_else(|| HookError::Parse("Invalid output header".to_string()))?;

        let output_strings_offset = { output_header.strings_offset } as usize;

        let parser = OutputParser::new(output_data, output_strings_offset)?;
        let entities: Result<Vec<ParsedEntity>, HookError> = parser.entities().collect();
        debug!(
            entity_count = entities.as_ref().map_or(0, std::vec::Vec::len),
            "output parse complete"
        );
        entities
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_executor_creation() {
        let executor = ScriptExecutor::try_new().expect("baseline engine");
        assert!(
            executor
                .host
                .module_cache()
                .is_empty()
                .expect("cache lock must not be poisoned in fresh executor")
        );
    }
}