/home/runner/work/lyquor/lyquor/platform/hosting/src/lvm.rs

Source
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use async_trait::async_trait;
use lyquor_api::{
    anyhow,
    call::CallParams,
    interface::{
        FetchOracleInfo, GetAddressByEd25519, GetAddressByEd25519Service, GetEd25519ByAddress,
        GetEd25519ByAddressService, GetEthContractAddr, GetEthContractAddrService, InterCallService, OnInterCall,
        OracleInfoService, SubmitCall, SubmitService,
    },
    store::{KVStore, KVStoreError, StoreFuture},
};
use lyquor_crypto as crypto;
use lyquor_primitives::oracle::OracleTarget;
use lyquor_primitives::{
    Address, Bytes, ConsoleSink, LyquidID, LyquidNumber, NodeID, SequenceBackendID, encode_object,
};
use lyquor_upc::{CallHeader, MulticastContext, RemoteObj};
use lyquor_vm::{RuntimeEnv, lyquid};
use thiserror::Error;
use tokio::sync::mpsc;
use tower::{ServiceExt, util::BoxCloneSyncService};

/// Per-Lyquid stores required to construct a VM instance.
pub struct ProcessStores {
    pub vm: lyquor_vm::InstanceStores,
    pub progress: Arc<dyn KVStore>,
}

/// Factory that returns namespaced stores for a Lyquid process.
pub type ProcessStoreFactory = Arc<dyn Fn(LyquidID) -> StoreFuture<Result<ProcessStores, KVStoreError>> + Send + Sync>;

/// Sequencing services exposed to hosted Lyquid VM host APIs.
#[derive(Clone)]
pub struct Sequencer {
    pub inter: InterCallService,
    pub submit: SubmitService,
    pub fetch_oracle_info: OracleInfoService,
}

/// Bartender lookup services exposed to hosted Lyquid VM host APIs.
#[derive(Clone, Debug)]
pub struct Bartender {
    pub get_address_by_ed25519: GetAddressByEd25519Service,
    pub get_ed25519_by_address: GetEd25519ByAddressService,
    pub get_eth_contract_address: GetEthContractAddrService,
    pub sequence_backend_id: SequenceBackendID,
}

/// Console output emitted by a hosted Lyquid.
#[derive(Debug, Clone)]
pub struct ConsoleOutput {
    pub sink: ConsoleSink,
    pub data: String,
}

/// Trigger registration or removal requested by a hosted Lyquid (see [`lyquor_primitives::TriggerMode`]).
#[derive(Debug, Clone)]
pub struct Trigger {
    pub group: String,
    pub method: String,
    pub input: Vec<u8>,
    pub version: LyquidNumber,
    pub mode: lyquor_primitives::TriggerMode,
}

/// Service used by VM host APIs to register or stop triggers.
pub type TriggerService = BoxCloneSyncService<Trigger, (), anyhow::Error>;
/// Shared VM instance type for hosting runtime environments.
pub type Instance = lyquor_vm::Instance<Env>;

// TODO: make those configurable (from Lyquid? and/or with some guard at Node level)
const HTTP_CONNECT_TIMEOUT: Duration = Duration::from_secs(5);
const HTTP_REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
const HTTP_POOL_IDLE_TIMEOUT: Duration = Duration::from_secs(90);
const HTTP_POOL_MAX_IDLE_PER_HOST: usize = 8;
const HTTP_TCP_KEEPALIVE: Duration = Duration::from_secs(60);

#[derive(Clone)]
struct HttpClient {
    client: Result<reqwest::Client, String>,
}

impl HttpClient {
    fn new() -> Self {
        let client = reqwest::Client::builder()
            .connect_timeout(HTTP_CONNECT_TIMEOUT)
            .timeout(HTTP_REQUEST_TIMEOUT)
            .pool_max_idle_per_host(HTTP_POOL_MAX_IDLE_PER_HOST)
            .pool_idle_timeout(Some(HTTP_POOL_IDLE_TIMEOUT))
            .tcp_keepalive(Some(HTTP_TCP_KEEPALIVE))
            .build()
            .map_err(|err| err0.to_string0());

        if let Err(ref err0) = client {
            tracing::warn!(error = %err, "failed to initialize instance http client");
        } else {
            tracing::trace!("initialized instance http client");
        }

        Self { client }
    }

    fn client(&self) -> Result<&reqwest::Client, lyquid::LyquidError> {
        self.client
            .as_ref()
            .map_err(|err| lyquid::LyquidError::LyquorRuntime(format!0("http client init failed: {err}")))
    }
}

#[derive(Clone)]
struct NodeEnv {
    upc: lyquor_upc::Requester<lyquor_vm::instance::Error>,
    node_id: NodeID,
    sequencer: Option<Sequencer>,
    sig_provider: Arc<crypto::SigProvider>,
    bartender: Option<Bartender>,
}

#[derive(Clone)]
struct LyquidEnv {
    lyquid_id: LyquidID,
    console: Option<mpsc::Sender<ConsoleOutput>>,
    http_client: HttpClient,
    trigger: Option<TriggerService>,
}

/// Runtime environment captured by hosted Lyquid VM instances.
#[derive(Clone)]
pub struct Env {
    node: NodeEnv,
    lyquid: LyquidEnv,
}

impl Env {
    fn sequencer(&self) -> Option<&Sequencer> {
        self.node.sequencer.as_ref()
    }
}

impl lyquor_vm::RuntimeEnv for Env {
    fn get_lyquid_id(&self) -> LyquidID {
        self.lyquid.lyquid_id
    }

    fn get_node_id(&self) -> NodeID {
        self.node.node_id
    }
}

/// Errors returned while constructing or wiring hosted VM runtime state.
#[derive(Debug, Error)]
pub enum VmRuntimeError {
    #[error("VM: {0}")]
    VM(#[from] lyquor_vm::Error),
    #[error("UPC: {0}")]
    UPC(#[from] lyquor_upc::Error<lyquor_vm::instance::Error>),
}

type HostAPIFunc = lyquor_vm::barrel::HostAPIFunc<lyquor_vm::instance::Host<Env>>;

lazy_static::lazy_static! {
    static ref LVM_CONSOLE: HostAPIFunc = lyquor_vm::host_api!((host: Host<Env>; sink: ConsoleSink, data: String) {
        // Silently drop the message if the output is not available.
        if let Some(console0) = &host.env.lyquid.console {
            console.send(ConsoleOutput {
                sink,
                data,
            }).await.ok();
        }
        Ok(())
    });

    static ref LVM_UPC: HostAPIFunc = lyquor_vm::host_api!((host: Host<Env>;
            target: LyquidID, group: Option<String>, method: String, input: Vec<u8>, client_params: Option<Bytes>, timeout_ms: Option<u64>) -> Vec<u8> {
        if host.category != lyquor_primitives::StateCategory::Instance {
            return Err(lyquid::LyquidError::LyquorRuntime(
                "`universal_procedural_call` only allowed in instance functions".to_string(),
            ));
        }

        tracing::trace!("Host handling UPC {:?} -> {:?} (group={:?}, method={}, input=<{} bytes>, client_params={:?})",
            host.env.lyquid.lyquid_id, target, group, method, input.len(), client_params);

        let number = if target == host.env.lyquid.lyquid_id {
            Some(*host.version.read().await)
        } else {
            None
        };
        let timeout = timeout_ms
            .map(Duration::from_millis)
            .unwrap_or_else(|| tokio::time::Duration::from_secs(lyquor_upc::message::UPC_PROCEDURE_TIMEOUT_DEFAULT));
        host.env.node.upc.call(CallHeader {
            lyquid: target,
            method,
            number,
            group,
            timeout,
            context: None,
        }, RemoteObj::from_encoded(input.into()), client_params, None).await
            .map_err(|e| lyquid::LyquidError::LyquorRuntime(format!0("UPC failed: {}", e)))
            .map(|r| Vec::from(r.to_encoded()))
    });

    static ref LVM_INTER_LYQUID_CALL: HostAPIFunc = lyquor_vm::host_api!((host: Host<Env>;
            callee: LyquidID, method: String, input: Vec<u8>) -> Vec<u8> {
        if host.category != lyquor_primitives::StateCategory::Network {
            return Err(lyquid::LyquidError::LyquorRuntime(
                "`inter_lyquid_call` only allowed in network functions".to_string(),
            ));
        }

        let pos = *host.chain_pos.read().await;
        if let Some(sys) = host.env.sequencer() {
            let caller = host.env.lyquid.lyquid_id;
            sys.inter
                .clone()
                .oneshot(OnInterCall {
                    origin: caller.into(), // TODO: pass on the real origin
                    caller,
                    callee,
                    method,
                    input,
                    pos,
                })
                .await
                .map_err(|e| lyquid::LyquidError::LyquorRuntime(format!0("error during service call: {:?}", e)))
        } else {
            Err(lyquid::LyquidError::LyquorRuntime("InterCallSys not available".into()))
        }
    });

    static ref LVM_SUBMIT_CALL: HostAPIFunc = lyquor_vm::host_api!((host: Host<Env>; call: CallParams, signed: bool) -> Vec<u8> {
        if host.category != lyquor_primitives::StateCategory::Instance {
            return Err(lyquid::LyquidError::LyquorRuntime(
                "`submit_call` only allowed in instance functions".to_string(),
            ));
        }

        let lyquid = host.env.lyquid.lyquid_id;
        if let Some(sys) = host.env.sequencer() {
            sys.submit.clone().oneshot(SubmitCall {
                lyquid,
                params: call,
                signed,
            }).await.map_err(|e|
                lyquid::LyquidError::LyquorRuntime(format!("submit call error: {:?}", e))
            )
        } else {
            Err(lyquid::LyquidError::LyquorRuntime("submit call recipient unavailable".into()))
        }
    });

    static ref LVM_SIGN: HostAPIFunc = lyquor_vm::host_api!((host: Host<Env>;
            msg: Bytes,
            cipher: lyquor_primitives::Cipher) -> lyquor_primitives::Signature {
        if host.category != lyquor_primitives::StateCategory::Instance {
            return Err(lyquid::LyquidError::LyquorRuntime(
                "`sign` only allowed in instance functions".to_string(),
            ));
        }

        let sig = host.env.node.sig_provider.sign(cipher, msg).await.unwrap_or_else(Bytes::new);
        Ok(sig)
    });

    static ref LVM_VERIFY: HostAPIFunc = lyquor_vm::host_api!((host: Host<Env>;
            msg: Bytes,
            cipher: lyquor_primitives::Cipher,
            sig: Bytes,
            pubkey: Bytes) -> bool {
        Ok(host.env.node.sig_provider.verify(cipher, msg, sig, pubkey).await)
    });

    static ref LVM_RNG: HostAPIFunc = lyquor_vm::host_api!((host: Host<Env>; length: usize) -> Vec<u8> {
        if host.category != lyquor_primitives::StateCategory::Instance {
            return Err(lyquid::LyquidError::LyquorRuntime(
                "random_bytes only allowed in instance functions (for now)".to_string(),
            ));
        }

        let len = if length > 256 {
            256
        } else {
            length
        };
        use rand::Rng;
        let mut bytes = vec![0u8; len];
        rand::rng().fill_bytes(&mut bytes);
        Ok(bytes)
    });

    static ref LVM_GET_TIME: HostAPIFunc = lyquor_vm::host_api!((host: Host<Env>;) -> u64 {
        if host.category != lyquor_primitives::StateCategory::Instance {
            return Err(lyquid::LyquidError::LyquorRuntime(
                "`systime` only allowed in instance functions".to_string(),
            ));
        }

        let now = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map_err(|e| lyquid::LyquidError::LyquorRuntime(format!("System time before unix epoch: {e}.")))?;
        u64::try_from(now.as_millis())
            .map_err(|_| lyquid::LyquidError::LyquorRuntime("Timestamp overflow.".to_string()))
    });

    static ref LVM_HTTP_REQUEST: HostAPIFunc = lyquor_vm::host_api!((host: Host<Env>;
            request: lyquid::http::Request,
            options: Option<lyquid::http::RequestOptions>) -> lyquid::http::Response {
        if host.category != lyquor_primitives::StateCategory::Instance {
            return Err(lyquid::LyquidError::LyquorRuntime(
                "`http_request` only allowed in instance functions".to_string(),
            ));
        }

        let lyquid::http::Request {
            method,
            url,
            headers,
            body,
        } = request;

        let method = match method {
            lyquid::http::Method::Get => reqwest::Method::GET,
            lyquid::http::Method::Head => reqwest::Method::HEAD,
            lyquid::http::Method::Post => reqwest::Method::POST,
            lyquid::http::Method::Put => reqwest::Method::PUT,
            lyquid::http::Method::Delete => reqwest::Method::DELETE,
            lyquid::http::Method::Connect => reqwest::Method::CONNECT,
            lyquid::http::Method::Options => reqwest::Method::OPTIONS,
            lyquid::http::Method::Trace => reqwest::Method::TRACE,
            lyquid::http::Method::Patch => reqwest::Method::PATCH,
            lyquid::http::Method::Other(method) => reqwest::Method::from_bytes(method.as_bytes()).map_err(|_| {
                lyquid::LyquidError::LyquorRuntime(format!("invalid http method: {method}"))
            })?,
        };

        let http_client = host.env.lyquid.http_client.clone();
        tracing::debug!(method = %method, url = %url, "dispatching http_request");

        let client = http_client.client()?0;
        let mut req = client.request(method, url);
        if let Some(options0) = options
            && let Some(timeout_ms) = options.timeout_ms
        {
            req = req.timeout(Duration::from_millis(timeout_ms));
        }
        for header0 in headers {
            let name = reqwest::header::HeaderName::from_bytes(header.name.as_bytes()).map_err(|e| {
                lyquid::LyquidError::LyquorRuntime(format!("invalid http header name: {e}"))
            })?;
            let value = reqwest::header::HeaderValue::from_bytes(&header.value).map_err(|e| {
                lyquid::LyquidError::LyquorRuntime(format!("invalid http header value: {e}"))
            })?;
            req = req.header(name, value);
        }
        if let Some(body0) = body {
            req = req.body(body);
        }

        let response = req
            .send()
            .await
            .map_err(|e| lyquid::LyquidError::LyquorRuntime(format!0("http request failed: {e}")))?0;
        let status = response.status().as_u16();
        let headers = response
            .headers()
            .iter()
            .map(|(name, value)| lyquid::http::Header {
                name: name.as_str().to_string(),
                value: value.as_bytes().to_vec(),
            })
            .collect();
        let body = response
            .bytes()
            .await
            .map_err(|e| lyquid::LyquidError::LyquorRuntime(format!0("http body read failed: {e}")))?0;

        Ok(lyquid::http::Response {
            status,
            headers,
            body: body.to_vec(),
        })
    });

    static ref LVM_GET_ED25519_QXY: HostAPIFunc = lyquor_vm::host_api!((host: Host<Env>; pubkey: [u8; 32]) -> (lyquor_primitives::U256, lyquor_primitives::U256) {
        let ret = crypto::ed25519::SCLPubkey::new(pubkey);
        Ok((ret.qx, ret.qy))
    });

    static ref LVM_GET_ADDRESS_BY_ED25519: HostAPIFunc = lyquor_vm::host_api!((host: Host<Env>; pubkey: [u8; 32]) -> Option<Address> {
        // NOTE: The result of this function is deterministic for network function, because we use the
        // same chain position here for the address retrieval.
        let pos = *host.chain_pos.read().await;
        host.env.node.bartender
            .as_ref()
            .ok_or(lyquid::LyquidError::LyquorRuntime("GetAddressByEd25519 function not supported.".into()))?
            .get_address_by_ed25519.clone()
            .oneshot(GetAddressByEd25519 {
                pos,
                id: pubkey.into(),
            })
            .await
            .map_err(|e| lyquid::LyquidError::LyquorRuntime(format!("GetAddressByEd25519 failed: {}", e)))
    });

    static ref LVM_GET_ED25519_BY_ADDRESS: HostAPIFunc = lyquor_vm::host_api!((host: Host<Env>; address: Address) -> Option<NodeID> {
        let pos = *host.chain_pos.read().await;
        host.env.node.bartender
            .as_ref()
            .ok_or(lyquid::LyquidError::LyquorRuntime("GetEd25519ByAddress function not supported.".into()))?
            .get_ed25519_by_address.clone()
            .oneshot(GetEd25519ByAddress {
                pos,
                address,
            })
            .await
            .map_err(|e| lyquid::LyquidError::LyquorRuntime(format!("GetEd25519ByAddress failed: {}", e)))
    });

    static ref LVM_ETH_CONTRACT: HostAPIFunc = lyquor_vm::host_api!((host: Host<Env>;) -> Option<Address> {
        // NOTE: The result of this function is deterministic for network function, because we use the
        // same chain position here for the address retrieval.
        let pos = *host.chain_pos.read().await;
        let id = host.env.get_lyquid_id();
        host.env.node.bartender
            .as_ref()
            .ok_or(lyquid::LyquidError::LyquorRuntime("eth_contract function not supported.".into()))?0
            .get_eth_contract_address
            .clone()
            .oneshot(GetEthContractAddr { pos, id })
            .await
            .map_err(|e| lyquid::LyquidError::LyquorRuntime(format!0("GetEthContractAddr failed: {}", e)))
    });

    static ref LVM_SEQUENCE_BACKEND_ID: HostAPIFunc = lyquor_vm::host_api!((host: Host<Env>;) -> SequenceBackendID {
        host.env
            .node
            .bartender
            .as_ref()
            .map(|b| b.sequence_backend_id)
            .ok_or(lyquid::LyquidError::LyquorRuntime("sequence_backend_id function not supported.".into()))
    });

    static ref LVM_FETCH_ORACLE_INFO: HostAPIFunc = lyquor_vm::host_api!((host: Host<Env>;
            topic: String,
            target: OracleTarget,
            full_config: bool) -> Option<lyquor_primitives::oracle::OracleEpochInfo> {
        if host.category != lyquor_primitives::StateCategory::Instance {
            return Err(lyquid::LyquidError::LyquorRuntime(
                "`fetch_oracle_info` only allowed in instance functions".to_string(),
            ));
        }
        let local_seq_id = host
            .env
            .node
            .bartender
            .as_ref()
            .ok_or(lyquid::LyquidError::LyquorRuntime("fetch_oracle_info not supported.".into()))?
            .sequence_backend_id;
        if target.seq_id != local_seq_id {
            return Ok(None);
        }

        let query = FetchOracleInfo {
            pos: *host.chain_pos.read().await,
            topic,
            target,
            full_config,
        };
        // NOTE: Query routing is currently local-only and does not dispatch by
        // `query.target.seq_id`. Cross-backend settlement should route by seq_id.
        host.env
            .sequencer()
            .ok_or(lyquid::LyquidError::LyquorRuntime(
                "fetch_oracle_info service unavailable.".into(),
            ))?
            .fetch_oracle_info
            .clone()
            .oneshot(query)
            .await
            .map_err(|e| lyquid::LyquidError::LyquorRuntime(format!("FetchOracleInfo failed: {}", e)))
    });

    static ref LVM_TRIGGER: HostAPIFunc = lyquor_vm::host_api!((host: Host<Env>; group: String, method: String, input: Vec<u8>, mode: lyquor_primitives::TriggerMode) -> () {
        // NOTE: Network and instance functions can both trigger instance functions directly, but
        // commit-mode triggers are only valid from network functions. This function should not
        // return anything.
        if matches!(mode, lyquor_primitives::TriggerMode::Commit)
            && host.category != lyquor_primitives::StateCategory::Network
        {
            return Err(lyquid::LyquidError::LyquorRuntime(
                "`TriggerMode::Commit` only allowed in network functions".to_string(),
            ));
        }

        let version = *host.version.read().await;
        host.env.lyquid.trigger
            .as_ref()
            .ok_or(lyquid::LyquidError::LyquorRuntime("`trigger` not supported.".into()))?
            .clone()
            .oneshot(Trigger {
                    group,
                    method,
                    input,
                    version,
                    mode,
                })
            .await
            .map_err(|_| lyquid::LyquidError::LyquorRuntime("Trigger failed: trigger queue is closed".into()))?;
        Ok(())
    });
}

struct Endpoint {
    instance: Instance,
}

impl std::fmt::Debug for Endpoint {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let id = self.instance.id();
        f.debug_struct("Endpoint").field("instance", &id).finish()
    }
}

#[derive(Clone)]
struct UpcCacheEntry {
    barrel: Arc<lyquor_vm::instance::Barrel<Env>>,
    cache_ptr: Option<lyquid::upc::CachePtr>,
}

#[async_trait]
impl lyquor_upc::Caller for Endpoint {
    type Error = lyquor_vm::instance::Error;

    #[tracing::instrument(level = "trace", skip(context))]
    async fn on_prepare(
        &self, header: &mut CallHeader, client_params: Bytes, context: &MulticastContext<Self::Error>,
    ) -> Result<Vec<NodeID>, Self::Error> {
        let input = lyquid::upc::PrepareInput {
            client_params: client_params.clone(),
        };
        let barrel = {
            let number = match header.number {
                Some(n) => n,
                // NOTE: Only execute latest_number() if not given.
                None => self.instance.latest_number().await,
            };
            // Write back exact lyquor number to header
            header.number = Some(number);
            self.instance
                .get_instance_barrel(number)
                .await
                .map_err(|e| lyquor_vm::instance::Error::Setup(e.into()))?
        };

        let call_params = CallParams::builder()
            .caller(Address::ZERO)
            .method(header.method.clone())
            .input(encode_object(&input).into())
            .group({
                let mut group = lyquor_primitives::GROUP_UPC_PREPARE.to_string();
                if let Some(ref suffix) = header.group {
                    group.push_str("::");
                    group.push_str(suffix);
                }
                group
            })
            .build();
        let call = barrel.call_func_decoded(
            call_params,
            Some((
                lyquor_vm::scheduler::RunOptions::new(lyquor_vm::scheduler::RunSource::UpcPrepare),
                self.instance.scheduler(),
            )),
        );
        let result: Result<lyquid::upc::PrepareOutput, Self::Error> = call.await.map_err(|e| e.into());
        let result = result?;

        {
            let mut cache = context.cache.lock().await;
            *cache = Some(Box::new(UpcCacheEntry {
                barrel,
                cache_ptr: result.cache,
            }));
        }

        Ok(result.result)
    }

    #[tracing::instrument(level = "trace", skip(context, returned))]
    async fn on_return(
        &self, header: CallHeader, from: NodeID, context: &MulticastContext<Self::Error>,
        returned: Result<RemoteObj, lyquor_upc::Error<Self::Error>>,
    ) -> Result<Option<RemoteObj>, lyquor_upc::Error<Self::Error>> {
        let id = context.id;
        let returned = returned?;

        let mut cache = context.cache.lock().await;
        if cache.is_none() {
            let number = match header.number {
                Some(n) => n,
                // NOTE: Only execute latest_number() if not given.
                None => self.instance.latest_number().await,
            };
            *cache = Some(Box::new(UpcCacheEntry {
                barrel: self
                    .instance
                    .get_instance_barrel(number)
                    .await
                    .map_err(lyquor_upc::Error::Call)?,
                cache_ptr: None,
            }));
        }
        let cache_entry = cache
            .as_mut()
            .unwrap()
            .as_mut()
            .downcast_mut::<UpcCacheEntry>()
            .unwrap();

        let input = lyquid::upc::ResponseInput {
            from,
            id,
            returned: returned.to_encoded().into(),
            cache: cache_entry.cache_ptr,
        };
        let call_params = CallParams::builder()
            .caller(Address::ZERO)
            .method(header.method)
            .input(encode_object(&input).into())
            .group({
                let mut group = lyquor_primitives::GROUP_UPC_RESP.to_string();
                if let Some(ref suffix) = header.group {
                    group.push_str("::");
                    group.push_str(suffix);
                }
                group
            })
            .build();
        let call = cache_entry.barrel.call_func_decoded(
            call_params,
            Some((
                lyquor_vm::scheduler::RunOptions::new(lyquor_vm::scheduler::RunSource::UpcResponse),
                self.instance.scheduler(),
            )),
        );
        let result: Result<lyquid::upc::ResponseOutput, lyquor_vm::instance::Error> =
            call.await.map_err(lyquor_vm::instance::Error::Barrel);
        let result = match result {
            Err(lyquor_vm::instance::Error::Barrel(lyquor_vm::barrel::Error::FuncNotFound)) => {
                // by default if upc_response event handler is not implemented, it will just
                // return the first seen result
                return Ok(Some(returned));
            }
            r => r,
        };
        let result = result.map_err(lyquor_upc::Error::Call)?;

        Ok(match result {
            lyquid::upc::ResponseOutput::Continue(cache) => {
                cache_entry.cache_ptr = cache;
                None
            }
            lyquid::upc::ResponseOutput::Return(result) => {
                *cache = None;
                Some(RemoteObj::from_encoded(result.into()))
            }
        })
    }
}

#[async_trait]
impl lyquor_upc::Callee for Endpoint {
    type Error = lyquor_vm::instance::Error;

    #[tracing::instrument(level = "trace", skip(input))]
    async fn on_call(&self, header: CallHeader, from: NodeID, input: RemoteObj) -> Result<RemoteObj, Self::Error> {
        let input = encode_object(&lyquid::upc::RequestInput {
            from,
            id: 0,
            input: input.as_encoded().unwrap().into(),
        })
        .into();
        let call = {
            // TODO: Now, header should always have LYQNUM present. However this leads to a problem
            // which the LYQNUM may not be a valid one on callee(which is me here). For now let's
            // check the existance of this number and if not let's go with the latest.
            let version = match header.number {
                Some(n) => n,
                // NOTE: Only execute latest_number() if not given.
                None => self.instance.latest_number().await,
            };
            let number = if self.instance.check_version(version).await {
                version
            } else {
                self.instance.latest_number().await
            };

            self.instance
                .call_instance_func(
                    number,
                    CallParams::builder()
                        .caller(Address::ZERO)
                        .method(header.method)
                        .input(input)
                        .group({
                            let mut group = lyquor_primitives::GROUP_UPC_REQ.to_string();
                            if let Some(ref suffix) = header.group {
                                group.push_str("::");
                                group.push_str(suffix);
                            }
                            group
                        })
                        .build(),
                    lyquor_vm::scheduler::RunOptions::new(lyquor_vm::scheduler::RunSource::UpcRequest),
                )
                .await
        };
        let result: lyquid::upc::RequestOutput = call.await?;
        Ok(RemoteObj::from_encoded(result.into()))
    }
}

/// Hosting-owned VM runtime plus UPC registration state.
pub struct VmRuntime {
    vm_engine: lyquor_vm::VmEngine,
    upc: lyquor_upc::UPC<lyquor_vm::instance::Error>,
    sequencer: Option<Sequencer>,
}

impl VmRuntime {
    /// Create a VM runtime bound to a network hub and optional sequencer services.
    pub fn new(
        runtime: tokio::runtime::Handle, network: Arc<lyquor_net::hub::Hub>, sequencer: Option<Sequencer>,
    ) -> Result<Self, VmRuntimeError> {
        Ok(Self {
            vm_engine: lyquor_vm::VmEngine::new(runtime)?0,
            upc: lyquor_upc::UPC::new(network),
            sequencer,
        })
    }

    /// Replace sequencer services exposed to subsequently created instances.
    pub fn set_sequencer(&mut self, sequencer: Option<Sequencer>) {
        self.sequencer = sequencer;
    }

    /// Add a remote UPC peer to the runtime.
    pub async fn add_remote(
        &mut self, node: NodeID, address: Option<String>,
    ) -> Result<(), lyquor_upc::Error<lyquor_vm::instance::Error>> {
        self.upc.add_remote(node, address).await
    }

    /// Return a UPC requester for hosted Lyquid calls.
    pub fn requester(&self) -> lyquor_upc::Requester<lyquor_vm::instance::Error> {
        self.upc.requester()
    }

    /// Create and register a hosted VM instance with all Lyquor host APIs.
    pub async fn add_instance(
        &mut self, id: LyquidID, stores: lyquor_vm::InstanceStores, console: Option<mpsc::Sender<ConsoleOutput>>,
        bartender: Option<Bartender>, trigger: Option<TriggerService>,
        image_source: Option<lyquor_vm::instance::ImageSource>,
    ) -> Result<Instance, VmRuntimeError> {
        let upc = self.upc.requester();
        let node_id = *self.upc.node_id();
        let sig_provider = Arc::new(crypto::SigProvider::new(self.upc.signing_key().as_ref().clone()));
        let http_client = HttpClient::new();
        let builder = self
            .vm_engine
            .instance_builder(
                id,
                stores,
                Env {
                    node: NodeEnv {
                        upc,
                        node_id,
                        sequencer: self.sequencer.clone(),
                        sig_provider,
                        bartender,
                    },
                    lyquid: LyquidEnv {
                        lyquid_id: id,
                        console,
                        http_client,
                        trigger,
                    },
                },
            )
            .await?0;
        let instance = builder
            .with_image_source(image_source)
            .host_api("console_output", LVM_CONSOLE.clone())
            .host_api("universal_procedural_call", LVM_UPC.clone())
            .host_api("inter_lyquid_call", LVM_INTER_LYQUID_CALL.clone())
            .host_api("submit_call", LVM_SUBMIT_CALL.clone())
            .host_api("sign", LVM_SIGN.clone())
            .host_api("verify", LVM_VERIFY.clone())
            .host_api("random_bytes", LVM_RNG.clone())
            .host_api("systime", LVM_GET_TIME.clone())
            .host_api("http_request", LVM_HTTP_REQUEST.clone())
            .host_api("get_ed25519_qxy", LVM_GET_ED25519_QXY.clone())
            .host_api("get_address_by_ed25519", LVM_GET_ADDRESS_BY_ED25519.clone())
            .host_api("get_ed25519_by_address", LVM_GET_ED25519_BY_ADDRESS.clone())
            .host_api("eth_contract", LVM_ETH_CONTRACT.clone())
            .host_api("sequence_backend_id", LVM_SEQUENCE_BACKEND_ID.clone())
            .host_api("fetch_oracle_info", LVM_FETCH_ORACLE_INFO.clone())
            .host_api("trigger", LVM_TRIGGER.clone())
            .build()
            .await
            .map_err(lyquor_vm::Error::from)?0;

        let instance = Arc::new(instance);
        self.upc
            .register_caller(
                &id,
                Endpoint {
                    instance: instance.clone(),
                },
            )
            .await?0;
        if let Err(err0) = self
            .upc
            .register_callee(
                &id,
                Endpoint {
                    instance: instance.clone(),
                },
            )
            .await
        {
            self.upc.deregister_caller(&id);
            return Err(err.into());
        }
        Ok(instance)
    }

    /// Deregister a hosted instance from UPC caller and callee tables.
    pub fn remove_instance(&mut self, id: &LyquidID) {
        self.upc.deregister_caller(id);
        self.upc.deregister_callee(id);
    }
}

Coverage Report

Created: 2026-06-13 03:29