/home/runner/work/lyquor/lyquor/platform/hosting/src/lvm.rs
Line | Count | Source |
1 | | use std::sync::Arc; |
2 | | use std::time::{Duration, SystemTime, UNIX_EPOCH}; |
3 | | |
4 | | use crate::egress::{Egress, EgressConfig}; |
5 | | use async_trait::async_trait; |
6 | | use lyquor_api::{ |
7 | | anyhow, |
8 | | call::CallParams, |
9 | | interface::{ |
10 | | FetchOracleInfo, GetAddressByEd25519, GetAddressByEd25519Service, GetEd25519ByAddress, |
11 | | GetEd25519ByAddressService, GetEthContractAddr, GetEthContractAddrService, InterCallService, OnInterCall, |
12 | | OracleInfoService, SubmitCall, SubmitService, |
13 | | }, |
14 | | store::{KVStore, KVStoreError, StoreFuture}, |
15 | | }; |
16 | | use lyquor_crypto as crypto; |
17 | | use lyquor_primitives::oracle::OracleTarget; |
18 | | use lyquor_primitives::{ |
19 | | Address, Bytes, ConsoleSink, LyquidID, LyquidNumber, NodeID, SequenceBackendID, encode_object, |
20 | | }; |
21 | | use lyquor_upc::{CallHeader, MulticastContext, RemoteObj}; |
22 | | use lyquor_vm::{RuntimeEnv, lyquid}; |
23 | | use thiserror::Error; |
24 | | use tokio::sync::mpsc; |
25 | | use tower::{ServiceExt, util::BoxCloneSyncService}; |
26 | | |
27 | | /// Per-Lyquid stores required to construct a VM instance. |
28 | | pub struct ProcessStores { |
29 | | pub vm: lyquor_vm::InstanceStores, |
30 | | pub progress: Arc<dyn KVStore>, |
31 | | } |
32 | | |
33 | | /// Factory that returns namespaced stores for a Lyquid process. |
34 | | pub type ProcessStoreFactory = Arc<dyn Fn(LyquidID) -> StoreFuture<Result<ProcessStores, KVStoreError>> + Send + Sync>; |
35 | | |
36 | | /// Sequencing services exposed to hosted Lyquid VM host APIs. |
37 | | #[derive(Clone)] |
38 | | pub struct Sequencer { |
39 | | pub inter: InterCallService, |
40 | | pub submit: SubmitService, |
41 | | pub fetch_oracle_info: OracleInfoService, |
42 | | } |
43 | | |
44 | | /// Bartender lookup services exposed to hosted Lyquid VM host APIs. |
45 | | #[derive(Clone, Debug)] |
46 | | pub struct Bartender { |
47 | | pub get_address_by_ed25519: GetAddressByEd25519Service, |
48 | | pub get_ed25519_by_address: GetEd25519ByAddressService, |
49 | | pub get_eth_contract_address: GetEthContractAddrService, |
50 | | } |
51 | | |
52 | | /// Console output emitted by a hosted Lyquid. |
53 | | #[derive(Debug, Clone)] |
54 | | pub struct ConsoleOutput { |
55 | | pub sink: ConsoleSink, |
56 | | pub data: String, |
57 | | } |
58 | | |
59 | | /// Trigger registration or removal requested by a hosted Lyquid (see [`lyquor_primitives::TriggerMode`]). |
60 | | #[derive(Debug, Clone)] |
61 | | pub struct Trigger { |
62 | | pub group: String, |
63 | | pub method: String, |
64 | | pub input: Vec<u8>, |
65 | | pub version: LyquidNumber, |
66 | | pub mode: lyquor_primitives::TriggerMode, |
67 | | } |
68 | | |
69 | | /// Service used by VM host APIs to register or stop triggers. |
70 | | pub type TriggerService = BoxCloneSyncService<Trigger, (), anyhow::Error>; |
71 | | /// Shared VM instance type for hosting runtime environments. |
72 | | pub type Instance = lyquor_vm::Instance<Env>; |
73 | | |
74 | | #[derive(Clone)] |
75 | | struct NodeEnv { |
76 | | upc: lyquor_upc::Requester<lyquor_vm::instance::Error>, |
77 | | node_id: NodeID, |
78 | | sequencer: Option<Sequencer>, |
79 | | sig_provider: Arc<crypto::SigProvider>, |
80 | | bartender: Option<Bartender>, |
81 | | // Chain-level constant available to every instance (including bartender, |
82 | | // which runs without the registry-backed `Bartender` service bundle). |
83 | | sequence_backend_id: SequenceBackendID, |
84 | | } |
85 | | |
86 | | #[derive(Clone)] |
87 | | struct LyquidEnv { |
88 | | lyquid_id: LyquidID, |
89 | | console: Option<mpsc::Sender<ConsoleOutput>>, |
90 | | egress: Egress, |
91 | | trigger: Option<TriggerService>, |
92 | | } |
93 | | |
94 | | /// Runtime environment captured by hosted Lyquid VM instances. |
95 | | #[derive(Clone)] |
96 | | pub struct Env { |
97 | | node: NodeEnv, |
98 | | lyquid: LyquidEnv, |
99 | | } |
100 | | |
101 | | impl Env { |
102 | 2 | fn sequencer(&self) -> Option<&Sequencer> { |
103 | 2 | self.node.sequencer.as_ref() |
104 | 2 | } |
105 | | } |
106 | | |
107 | | impl lyquor_vm::RuntimeEnv for Env { |
108 | 261 | fn get_lyquid_id(&self) -> LyquidID { |
109 | 261 | self.lyquid.lyquid_id |
110 | 261 | } |
111 | | |
112 | 173 | fn get_node_id(&self) -> NodeID { |
113 | 173 | self.node.node_id |
114 | 173 | } |
115 | | } |
116 | | |
117 | | /// Errors returned while constructing or wiring hosted VM runtime state. |
118 | | #[derive(Debug, Error)] |
119 | | pub enum VmRuntimeError { |
120 | | #[error("VM: {0}")] |
121 | | VM(#[from] lyquor_vm::Error), |
122 | | #[error("UPC: {0}")] |
123 | | UPC(#[from] lyquor_upc::Error<lyquor_vm::instance::Error>), |
124 | | } |
125 | | |
126 | | type HostAPIFunc = lyquor_vm::barrel::HostAPIFunc<lyquor_vm::instance::Host<Env>>; |
127 | | |
128 | | lazy_static::lazy_static! { |
129 | 158 | static ref LVM_CONSOLE: HostAPIFunc = lyquor_vm::host_api!((host: Host<Env>; sink: ConsoleSink, data: String) { |
130 | | // Silently drop the message if the output is not available. |
131 | 158 | if let Some(console0 ) = &host.env.lyquid.console { |
132 | 0 | console.send(ConsoleOutput { |
133 | 0 | sink, |
134 | 0 | data, |
135 | 0 | }).await.ok(); |
136 | 158 | } |
137 | 158 | Ok(()) |
138 | | }); |
139 | | |
140 | | static ref LVM_UPC: HostAPIFunc = lyquor_vm::host_api!((host: Host<Env>; |
141 | 7 | target: LyquidID, group: Option<String>, method: String, input: Vec<u8>, client_params: Option<Bytes>, timeout_ms: Option<u64>) -> Vec<u8> { |
142 | 7 | if host.category != lyquor_primitives::StateCategory::Instance { |
143 | 1 | return Err(lyquid::LyquidError::LyquorRuntime( |
144 | 1 | "`universal_procedural_call` only allowed in instance functions".to_string(), |
145 | 1 | )); |
146 | 6 | } |
147 | | |
148 | 6 | tracing::trace!("Host handling UPC {:?} -> {:?} (group={:?}, method={}, input=<{} bytes>, client_params={:?})", |
149 | 0 | host.env.lyquid.lyquid_id, target, group, method, input.len(), client_params); |
150 | | |
151 | 6 | let number = if target == host.env.lyquid.lyquid_id { |
152 | 3 | Some(*host.version.read().await) |
153 | | } else { |
154 | 3 | None |
155 | | }; |
156 | 6 | let timeout = timeout_ms.map_or_else(|| tokio::time::Duration::from_secs(lyquor_upc::message::UPC_PROCEDURE_TIMEOUT_DEFAULT), Duration::from_millis); |
157 | 6 | host.env.node.upc.call(CallHeader { |
158 | 6 | lyquid: target, |
159 | 6 | method, |
160 | 6 | number, |
161 | 6 | group, |
162 | 6 | timeout, |
163 | 6 | context: None, |
164 | 6 | }, RemoteObj::from_encoded(input.into()), client_params, None).await |
165 | 6 | .map_err(|e| lyquid::LyquidError::LyquorRuntime(format!0 ("UPC failed: {e}"))) |
166 | 6 | .map(|r| Vec::from(r.to_encoded())) |
167 | | }); |
168 | | |
169 | | static ref LVM_INTER_LYQUID_CALL: HostAPIFunc = lyquor_vm::host_api!((host: Host<Env>; |
170 | 2 | callee: LyquidID, method: String, input: Vec<u8>) -> Vec<u8> { |
171 | 2 | if host.category != lyquor_primitives::StateCategory::Network { |
172 | 0 | return Err(lyquid::LyquidError::LyquorRuntime( |
173 | 0 | "`inter_lyquid_call` only allowed in network functions".to_string(), |
174 | 0 | )); |
175 | 2 | } |
176 | | |
177 | 2 | let pos = *host.chain_pos.read().await; |
178 | 2 | if let Some(sys) = host.env.sequencer() { |
179 | 2 | let caller = host.env.lyquid.lyquid_id; |
180 | 2 | sys.inter |
181 | 2 | .clone() |
182 | 2 | .oneshot(OnInterCall { |
183 | 2 | origin: caller.into(), // TODO: pass on the real origin |
184 | 2 | caller, |
185 | 2 | callee, |
186 | 2 | method, |
187 | 2 | input, |
188 | 2 | pos, |
189 | 2 | }) |
190 | 2 | .await |
191 | 2 | .map_err(|e| lyquid::LyquidError::LyquorRuntime(format!0 ("error during service call: {e:?}"))) |
192 | | } else { |
193 | 0 | Err(lyquid::LyquidError::LyquorRuntime("InterCallSys not available".into())) |
194 | | } |
195 | | }); |
196 | | |
197 | 0 | static ref LVM_SUBMIT_CALL: HostAPIFunc = lyquor_vm::host_api!((host: Host<Env>; call: CallParams, signed: bool) -> Vec<u8> { |
198 | 0 | if host.category != lyquor_primitives::StateCategory::Instance { |
199 | 0 | return Err(lyquid::LyquidError::LyquorRuntime( |
200 | 0 | "`submit_call` only allowed in instance functions".to_string(), |
201 | 0 | )); |
202 | 0 | } |
203 | | |
204 | 0 | let lyquid = host.env.lyquid.lyquid_id; |
205 | 0 | if let Some(sys) = host.env.sequencer() { |
206 | 0 | sys.submit.clone().oneshot(SubmitCall { |
207 | 0 | lyquid, |
208 | 0 | params: call, |
209 | 0 | signed, |
210 | 0 | }).await.map_err(|e| |
211 | 0 | lyquid::LyquidError::LyquorRuntime(format!("submit call error: {e:?}")) |
212 | | ) |
213 | | } else { |
214 | 0 | Err(lyquid::LyquidError::LyquorRuntime("submit call recipient unavailable".into())) |
215 | | } |
216 | | }); |
217 | | |
218 | | static ref LVM_SIGN: HostAPIFunc = lyquor_vm::host_api!((host: Host<Env>; |
219 | | msg: Bytes, |
220 | 0 | cipher: lyquor_primitives::Cipher) -> lyquor_primitives::Signature { |
221 | 0 | if host.category != lyquor_primitives::StateCategory::Instance { |
222 | 0 | return Err(lyquid::LyquidError::LyquorRuntime( |
223 | 0 | "`sign` only allowed in instance functions".to_string(), |
224 | 0 | )); |
225 | 0 | } |
226 | | |
227 | 0 | let sig = host.env.node.sig_provider.sign(cipher, msg).await.unwrap_or_else(Bytes::new); |
228 | 0 | Ok(sig) |
229 | | }); |
230 | | |
231 | | static ref LVM_VERIFY: HostAPIFunc = lyquor_vm::host_api!((host: Host<Env>; |
232 | | msg: Bytes, |
233 | | cipher: lyquor_primitives::Cipher, |
234 | | sig: Bytes, |
235 | 0 | pubkey: Bytes) -> bool { |
236 | 0 | Ok(host.env.node.sig_provider.verify(cipher, msg, sig, pubkey).await) |
237 | | }); |
238 | | |
239 | 0 | static ref LVM_RNG: HostAPIFunc = lyquor_vm::host_api!((host: Host<Env>; length: usize) -> Vec<u8> { |
240 | 0 | if host.category != lyquor_primitives::StateCategory::Instance { |
241 | 0 | return Err(lyquid::LyquidError::LyquorRuntime( |
242 | 0 | "random_bytes only allowed in instance functions (for now)".to_string(), |
243 | 0 | )); |
244 | 0 | } |
245 | | |
246 | 0 | let len = if length > 256 { |
247 | 0 | 256 |
248 | | } else { |
249 | 0 | length |
250 | | }; |
251 | | use rand::Rng; |
252 | 0 | let mut bytes = vec![0u8; len]; |
253 | 0 | rand::rng().fill_bytes(&mut bytes); |
254 | 0 | Ok(bytes) |
255 | | }); |
256 | | |
257 | 0 | static ref LVM_GET_TIME: HostAPIFunc = lyquor_vm::host_api!((host: Host<Env>;) -> u64 { |
258 | 0 | if host.category != lyquor_primitives::StateCategory::Instance { |
259 | 0 | return Err(lyquid::LyquidError::LyquorRuntime( |
260 | 0 | "`systime` only allowed in instance functions".to_string(), |
261 | 0 | )); |
262 | 0 | } |
263 | | |
264 | 0 | let now = SystemTime::now() |
265 | 0 | .duration_since(UNIX_EPOCH) |
266 | 0 | .map_err(|e| lyquid::LyquidError::LyquorRuntime(format!("System time before unix epoch: {e}.")))?; |
267 | 0 | u64::try_from(now.as_millis()) |
268 | 0 | .map_err(|_| lyquid::LyquidError::LyquorRuntime("Timestamp overflow.".to_string())) |
269 | | }); |
270 | | |
271 | | static ref LVM_HTTP_REQUEST: HostAPIFunc = lyquor_vm::host_api!((host: Host<Env>; |
272 | | request: lyquid::http::Request, |
273 | 8 | options: Option<lyquid::http::RequestOptions>) -> lyquid::http::Response { |
274 | 8 | if host.category != lyquor_primitives::StateCategory::Instance { |
275 | 0 | return Err(lyquid::LyquidError::LyquorRuntime( |
276 | 0 | "`http_request` only allowed in instance functions".to_string(), |
277 | 0 | )); |
278 | 8 | } |
279 | | |
280 | 8 | host.env.lyquid.egress.clone().send(request, options).await |
281 | | }); |
282 | | |
283 | 0 | static ref LVM_GET_ED25519_QXY: HostAPIFunc = lyquor_vm::host_api!((host: Host<Env>; pubkey: [u8; 32]) -> (lyquor_primitives::U256, lyquor_primitives::U256) { |
284 | 0 | let ret = crypto::ed25519::SCLPubkey::new(pubkey); |
285 | 0 | Ok((ret.qx, ret.qy)) |
286 | | }); |
287 | | |
288 | 0 | static ref LVM_GET_ADDRESS_BY_ED25519: HostAPIFunc = lyquor_vm::host_api!((host: Host<Env>; pubkey: [u8; 32]) -> Option<Address> { |
289 | | // NOTE: The result of this function is deterministic for network function, because we use the |
290 | | // same chain position here for the address retrieval. |
291 | 0 | let pos = *host.chain_pos.read().await; |
292 | 0 | host.env.node.bartender |
293 | 0 | .as_ref() |
294 | 0 | .ok_or_else(|| lyquid::LyquidError::LyquorRuntime("GetAddressByEd25519 function not supported.".into()))? |
295 | 0 | .get_address_by_ed25519.clone() |
296 | 0 | .oneshot(GetAddressByEd25519 { |
297 | 0 | pos, |
298 | 0 | id: pubkey.into(), |
299 | 0 | }) |
300 | 0 | .await |
301 | 0 | .map_err(|e| lyquid::LyquidError::LyquorRuntime(format!("GetAddressByEd25519 failed: {e}"))) |
302 | | }); |
303 | | |
304 | 0 | static ref LVM_GET_ED25519_BY_ADDRESS: HostAPIFunc = lyquor_vm::host_api!((host: Host<Env>; address: Address) -> Option<NodeID> { |
305 | 0 | let pos = *host.chain_pos.read().await; |
306 | 0 | host.env.node.bartender |
307 | 0 | .as_ref() |
308 | 0 | .ok_or_else(|| lyquid::LyquidError::LyquorRuntime("GetEd25519ByAddress function not supported.".into()))? |
309 | 0 | .get_ed25519_by_address.clone() |
310 | 0 | .oneshot(GetEd25519ByAddress { |
311 | 0 | pos, |
312 | 0 | address, |
313 | 0 | }) |
314 | 0 | .await |
315 | 0 | .map_err(|e| lyquid::LyquidError::LyquorRuntime(format!("GetEd25519ByAddress failed: {e}"))) |
316 | | }); |
317 | | |
318 | 12 | static ref LVM_ETH_CONTRACT: HostAPIFunc = lyquor_vm::host_api!((host: Host<Env>;) -> Option<Address> { |
319 | | // NOTE: The result of this function is deterministic for network function, because we use the |
320 | | // same chain position here for the address retrieval. |
321 | 12 | let pos = *host.chain_pos.read().await; |
322 | 12 | let id = host.env.get_lyquid_id(); |
323 | 12 | host.env.node.bartender |
324 | 12 | .as_ref() |
325 | 12 | .ok_or_else(|| lyquid::LyquidError::LyquorRuntime("eth_contract function not supported."0 .into0 ()))?0 |
326 | | .get_eth_contract_address |
327 | 12 | .clone() |
328 | 12 | .oneshot(GetEthContractAddr { pos, id }) |
329 | 12 | .await |
330 | 12 | .map_err(|e| lyquid::LyquidError::LyquorRuntime(format!0 ("GetEthContractAddr failed: {e}"))) |
331 | | }); |
332 | | |
333 | 13 | static ref LVM_SEQUENCE_BACKEND_ID: HostAPIFunc = lyquor_vm::host_api!((host: Host<Env>;) -> SequenceBackendID { |
334 | 13 | Ok(host.env.node.sequence_backend_id) |
335 | | }); |
336 | | |
337 | | static ref LVM_FETCH_ORACLE_INFO: HostAPIFunc = lyquor_vm::host_api!((host: Host<Env>; |
338 | | topic: String, |
339 | | target: OracleTarget, |
340 | 0 | full_config: bool) -> Option<lyquor_primitives::oracle::OracleEpochInfo> { |
341 | 0 | if host.category != lyquor_primitives::StateCategory::Instance { |
342 | 0 | return Err(lyquid::LyquidError::LyquorRuntime( |
343 | 0 | "`fetch_oracle_info` only allowed in instance functions".to_string(), |
344 | 0 | )); |
345 | 0 | } |
346 | 0 | if target.seq_id != host.env.node.sequence_backend_id { |
347 | 0 | return Ok(None); |
348 | 0 | } |
349 | | |
350 | 0 | let query = FetchOracleInfo { |
351 | 0 | pos: *host.chain_pos.read().await, |
352 | 0 | topic, |
353 | 0 | target, |
354 | 0 | full_config, |
355 | | }; |
356 | | // NOTE: Query routing is currently local-only and does not dispatch by |
357 | | // `query.target.seq_id`. Cross-backend settlement should route by seq_id. |
358 | 0 | host.env |
359 | 0 | .sequencer() |
360 | 0 | .ok_or_else(|| lyquid::LyquidError::LyquorRuntime( |
361 | 0 | "fetch_oracle_info service unavailable.".into(), |
362 | 0 | ))? |
363 | | .fetch_oracle_info |
364 | 0 | .clone() |
365 | 0 | .oneshot(query) |
366 | 0 | .await |
367 | 0 | .map_err(|e| lyquid::LyquidError::LyquorRuntime(format!("FetchOracleInfo failed: {e}"))) |
368 | | }); |
369 | | |
370 | 0 | static ref LVM_TRIGGER: HostAPIFunc = lyquor_vm::host_api!((host: Host<Env>; group: String, method: String, input: Vec<u8>, mode: lyquor_primitives::TriggerMode) -> () { |
371 | | // NOTE: Network and instance functions can both trigger instance functions directly, but |
372 | | // commit-mode triggers are only valid from network functions. This function should not |
373 | | // return anything. |
374 | 0 | if matches!(mode, lyquor_primitives::TriggerMode::Commit) |
375 | 0 | && host.category != lyquor_primitives::StateCategory::Network |
376 | | { |
377 | 0 | return Err(lyquid::LyquidError::LyquorRuntime( |
378 | 0 | "`TriggerMode::Commit` only allowed in network functions".to_string(), |
379 | 0 | )); |
380 | 0 | } |
381 | | |
382 | 0 | let version = *host.version.read().await; |
383 | 0 | host.env.lyquid.trigger |
384 | 0 | .as_ref() |
385 | 0 | .ok_or_else(|| lyquid::LyquidError::LyquorRuntime("`trigger` not supported.".into()))? |
386 | 0 | .clone() |
387 | 0 | .oneshot(Trigger { |
388 | 0 | group, |
389 | 0 | method, |
390 | 0 | input, |
391 | 0 | version, |
392 | 0 | mode, |
393 | 0 | }) |
394 | 0 | .await |
395 | 0 | .map_err(|_| lyquid::LyquidError::LyquorRuntime("Trigger failed: trigger queue is closed".into()))?; |
396 | 0 | Ok(()) |
397 | | }); |
398 | | } |
399 | | |
400 | | struct Endpoint { |
401 | | instance: Instance, |
402 | | } |
403 | | |
404 | | impl std::fmt::Debug for Endpoint { |
405 | 0 | fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { |
406 | 0 | let id = self.instance.id(); |
407 | 0 | f.debug_struct("Endpoint").field("instance", &id).finish() |
408 | 0 | } |
409 | | } |
410 | | |
411 | | #[derive(Clone)] |
412 | | struct UpcCacheEntry { |
413 | | barrel: Arc<lyquor_vm::instance::Barrel<Env>>, |
414 | | cache_ptr: Option<lyquid::upc::CachePtr>, |
415 | | } |
416 | | |
417 | | #[async_trait] |
418 | | impl lyquor_upc::Caller for Endpoint { |
419 | | type Error = lyquor_vm::instance::Error; |
420 | | |
421 | | #[tracing::instrument(level = "trace", skip(context))] |
422 | | async fn on_prepare( |
423 | | &self, header: &mut CallHeader, client_params: Bytes, context: &MulticastContext<Self::Error>, |
424 | | ) -> Result<Vec<NodeID>, Self::Error> { |
425 | | let input = lyquid::upc::PrepareInput { |
426 | | client_params: client_params.clone(), |
427 | | }; |
428 | | let barrel = { |
429 | | let number = match header.number { |
430 | | Some(n) => n, |
431 | | // NOTE: Only execute latest_number() if not given. |
432 | | None => self.instance.latest_number().await, |
433 | | }; |
434 | | // Write back exact lyquor number to header |
435 | | header.number = Some(number); |
436 | | self.instance |
437 | | .get_instance_barrel(number) |
438 | | .await |
439 | 0 | .map_err(|e| lyquor_vm::instance::Error::Setup(e.into()))? |
440 | | }; |
441 | | |
442 | | let call_params = CallParams::builder() |
443 | | .caller(Address::ZERO) |
444 | | .method(header.method.clone()) |
445 | | .input(encode_object(&input).into()) |
446 | | .group({ |
447 | | let mut group = lyquor_primitives::GROUP_UPC_PREPARE.to_string(); |
448 | | if let Some(ref suffix) = header.group { |
449 | | group.push_str("::"); |
450 | | group.push_str(suffix); |
451 | | } |
452 | | group |
453 | | }) |
454 | | .build(); |
455 | | let call = barrel.call_func_decoded( |
456 | | call_params, |
457 | | Some(( |
458 | | lyquor_vm::scheduler::RunOptions::new(lyquor_vm::scheduler::RunSource::UpcPrepare), |
459 | | self.instance.scheduler(), |
460 | | )), |
461 | | ); |
462 | | let result: Result<lyquid::upc::PrepareOutput, Self::Error> = call.await.map_err(std::convert::Into::into); |
463 | | let result = result?; |
464 | | |
465 | | { |
466 | | let mut cache = context.cache.lock().await; |
467 | | *cache = Some(Box::new(UpcCacheEntry { |
468 | | barrel, |
469 | | cache_ptr: result.cache, |
470 | | })); |
471 | | } |
472 | | |
473 | | Ok(result.result) |
474 | | } |
475 | | |
476 | | #[tracing::instrument(level = "trace", skip(context, returned))] |
477 | | async fn on_return( |
478 | | &self, header: CallHeader, from: NodeID, context: &MulticastContext<Self::Error>, |
479 | | returned: Result<RemoteObj, lyquor_upc::Error<Self::Error>>, |
480 | | ) -> Result<Option<RemoteObj>, lyquor_upc::Error<Self::Error>> { |
481 | | let id = context.id; |
482 | | let returned = returned?; |
483 | | |
484 | | let mut cache = context.cache.lock().await; |
485 | | if cache.is_none() { |
486 | | let number = match header.number { |
487 | | Some(n) => n, |
488 | | // NOTE: Only execute latest_number() if not given. |
489 | | None => self.instance.latest_number().await, |
490 | | }; |
491 | | *cache = Some(Box::new(UpcCacheEntry { |
492 | | barrel: self |
493 | | .instance |
494 | | .get_instance_barrel(number) |
495 | | .await |
496 | | .map_err(lyquor_upc::Error::Call)?, |
497 | | cache_ptr: None, |
498 | | })); |
499 | | } |
500 | | let cache_entry = cache |
501 | | .as_mut() |
502 | | .unwrap() |
503 | | .as_mut() |
504 | | .downcast_mut::<UpcCacheEntry>() |
505 | | .unwrap(); |
506 | | |
507 | | let input = lyquid::upc::ResponseInput { |
508 | | from, |
509 | | id, |
510 | | returned: returned.to_encoded().into(), |
511 | | cache: cache_entry.cache_ptr, |
512 | | }; |
513 | | let call_params = CallParams::builder() |
514 | | .caller(Address::ZERO) |
515 | | .method(header.method) |
516 | | .input(encode_object(&input).into()) |
517 | | .group({ |
518 | | let mut group = lyquor_primitives::GROUP_UPC_RESP.to_string(); |
519 | | if let Some(ref suffix) = header.group { |
520 | | group.push_str("::"); |
521 | | group.push_str(suffix); |
522 | | } |
523 | | group |
524 | | }) |
525 | | .build(); |
526 | | let call = cache_entry.barrel.call_func_decoded( |
527 | | call_params, |
528 | | Some(( |
529 | | lyquor_vm::scheduler::RunOptions::new(lyquor_vm::scheduler::RunSource::UpcResponse), |
530 | | self.instance.scheduler(), |
531 | | )), |
532 | | ); |
533 | | let result: Result<lyquid::upc::ResponseOutput, lyquor_vm::instance::Error> = |
534 | | call.await.map_err(lyquor_vm::instance::Error::Barrel); |
535 | | let result = match result { |
536 | | Err(lyquor_vm::instance::Error::Barrel(lyquor_vm::barrel::Error::FuncNotFound)) => { |
537 | | // by default if upc_response event handler is not implemented, it will just |
538 | | // return the first seen result |
539 | | return Ok(Some(returned)); |
540 | | } |
541 | | r => r, |
542 | | }; |
543 | | let result = result.map_err(lyquor_upc::Error::Call)?; |
544 | | |
545 | | Ok(match result { |
546 | | lyquid::upc::ResponseOutput::Continue(cache) => { |
547 | | cache_entry.cache_ptr = cache; |
548 | | None |
549 | | } |
550 | | lyquid::upc::ResponseOutput::Return(result) => { |
551 | | *cache = None; |
552 | | Some(RemoteObj::from_encoded(result.into())) |
553 | | } |
554 | | }) |
555 | | } |
556 | | } |
557 | | |
558 | | #[async_trait] |
559 | | impl lyquor_upc::Callee for Endpoint { |
560 | | type Error = lyquor_vm::instance::Error; |
561 | | |
562 | | #[tracing::instrument(level = "trace", skip(input))] |
563 | | async fn on_call(&self, header: CallHeader, from: NodeID, input: RemoteObj) -> Result<RemoteObj, Self::Error> { |
564 | | let input = encode_object(&lyquid::upc::RequestInput { |
565 | | from, |
566 | | id: 0, |
567 | | input: input.as_encoded().unwrap().into(), |
568 | | }) |
569 | | .into(); |
570 | | let call = { |
571 | | // TODO: Now, header should always have LYQNUM present. However this leads to a problem |
572 | | // which the LYQNUM may not be a valid one on callee(which is me here). For now let's |
573 | | // check the existance of this number and if not let's go with the latest. |
574 | | let version = match header.number { |
575 | | Some(n) => n, |
576 | | // NOTE: Only execute latest_number() if not given. |
577 | | None => self.instance.latest_number().await, |
578 | | }; |
579 | | let number = if self.instance.check_version(version).await { |
580 | | version |
581 | | } else { |
582 | | self.instance.latest_number().await |
583 | | }; |
584 | | |
585 | | self.instance |
586 | | .call_instance_func( |
587 | | number, |
588 | | CallParams::builder() |
589 | | .caller(Address::ZERO) |
590 | | .method(header.method) |
591 | | .input(input) |
592 | | .group({ |
593 | | let mut group = lyquor_primitives::GROUP_UPC_REQ.to_string(); |
594 | | if let Some(ref suffix) = header.group { |
595 | | group.push_str("::"); |
596 | | group.push_str(suffix); |
597 | | } |
598 | | group |
599 | | }) |
600 | | .build(), |
601 | | lyquor_vm::scheduler::RunOptions::new(lyquor_vm::scheduler::RunSource::UpcRequest), |
602 | | ) |
603 | | .await |
604 | | }; |
605 | | let result: lyquid::upc::RequestOutput = call.await?; |
606 | | Ok(RemoteObj::from_encoded(result.into())) |
607 | | } |
608 | | } |
609 | | |
610 | | /// Hosting-owned VM runtime plus UPC registration state. |
611 | | pub struct VmRuntime { |
612 | | vm_engine: lyquor_vm::VmEngine, |
613 | | upc: lyquor_upc::UPC<lyquor_vm::instance::Error>, |
614 | | sequencer: Option<Sequencer>, |
615 | | egress_config: EgressConfig, |
616 | | sequence_backend_id: SequenceBackendID, |
617 | | } |
618 | | |
619 | | impl VmRuntime { |
620 | | /// Create a VM runtime bound to a network hub and optional sequencer services. |
621 | 35 | pub fn new( |
622 | 35 | runtime: tokio::runtime::Handle, network: Arc<lyquor_net::hub::Hub>, sequencer: Option<Sequencer>, |
623 | 35 | egress_config: EgressConfig, sequence_backend_id: SequenceBackendID, |
624 | 35 | ) -> Result<Self, VmRuntimeError> { |
625 | | Ok(Self { |
626 | 35 | vm_engine: lyquor_vm::VmEngine::new(runtime)?0 , |
627 | 35 | upc: lyquor_upc::UPC::new(network), |
628 | 35 | sequencer, |
629 | 35 | egress_config, |
630 | 35 | sequence_backend_id, |
631 | | }) |
632 | 35 | } |
633 | | |
634 | | /// Replace sequencer services exposed to subsequently created instances. |
635 | 0 | pub fn set_sequencer(&mut self, sequencer: Option<Sequencer>) { |
636 | 0 | self.sequencer = sequencer; |
637 | 0 | } |
638 | | |
639 | | /// Return a UPC requester for hosted Lyquid calls. |
640 | 23 | pub fn requester(&self) -> lyquor_upc::Requester<lyquor_vm::instance::Error> { |
641 | 23 | self.upc.requester() |
642 | 23 | } |
643 | | |
644 | | /// Create and register a hosted VM instance with all Lyquor host APIs. |
645 | 42 | pub async fn add_instance( |
646 | 42 | &mut self, id: LyquidID, stores: lyquor_vm::InstanceStores, console: Option<mpsc::Sender<ConsoleOutput>>, |
647 | 42 | bartender: Option<Bartender>, trigger: Option<TriggerService>, |
648 | 42 | image_source: Option<lyquor_vm::instance::ImageSource>, |
649 | 42 | ) -> Result<Instance, VmRuntimeError> { |
650 | 42 | let upc = self.upc.requester(); |
651 | 42 | let node_id = *self.upc.node_id(); |
652 | 42 | let sig_provider = Arc::new(crypto::SigProvider::new(self.upc.signing_key().as_ref().clone())); |
653 | 42 | let egress = Egress::new(self.egress_config.clone()); |
654 | 42 | let builder = self |
655 | 42 | .vm_engine |
656 | 42 | .instance_builder( |
657 | 42 | id, |
658 | 42 | stores, |
659 | 42 | Env { |
660 | 42 | node: NodeEnv { |
661 | 42 | upc, |
662 | 42 | node_id, |
663 | 42 | sequencer: self.sequencer.clone(), |
664 | 42 | sig_provider, |
665 | 42 | bartender, |
666 | 42 | sequence_backend_id: self.sequence_backend_id, |
667 | 42 | }, |
668 | 42 | lyquid: LyquidEnv { |
669 | 42 | lyquid_id: id, |
670 | 42 | console, |
671 | 42 | egress, |
672 | 42 | trigger, |
673 | 42 | }, |
674 | 42 | }, |
675 | 42 | ) |
676 | 42 | .await?0 ; |
677 | 42 | let instance = builder |
678 | 42 | .with_image_source(image_source) |
679 | 42 | .host_api("console_output", LVM_CONSOLE.clone()) |
680 | 42 | .host_api("universal_procedural_call", LVM_UPC.clone()) |
681 | 42 | .host_api("inter_lyquid_call", LVM_INTER_LYQUID_CALL.clone()) |
682 | 42 | .host_api("submit_call", LVM_SUBMIT_CALL.clone()) |
683 | 42 | .host_api("sign", LVM_SIGN.clone()) |
684 | 42 | .host_api("verify", LVM_VERIFY.clone()) |
685 | 42 | .host_api("random_bytes", LVM_RNG.clone()) |
686 | 42 | .host_api("systime", LVM_GET_TIME.clone()) |
687 | 42 | .host_api("http_request", LVM_HTTP_REQUEST.clone()) |
688 | 42 | .host_api("get_ed25519_qxy", LVM_GET_ED25519_QXY.clone()) |
689 | 42 | .host_api("get_address_by_ed25519", LVM_GET_ADDRESS_BY_ED25519.clone()) |
690 | 42 | .host_api("get_ed25519_by_address", LVM_GET_ED25519_BY_ADDRESS.clone()) |
691 | 42 | .host_api("eth_contract", LVM_ETH_CONTRACT.clone()) |
692 | 42 | .host_api("sequence_backend_id", LVM_SEQUENCE_BACKEND_ID.clone()) |
693 | 42 | .host_api("fetch_oracle_info", LVM_FETCH_ORACLE_INFO.clone()) |
694 | 42 | .host_api("trigger", LVM_TRIGGER.clone()) |
695 | 42 | .build() |
696 | 42 | .await |
697 | 42 | .map_err(lyquor_vm::Error::from)?0 ; |
698 | | |
699 | 42 | let instance = Arc::new(instance); |
700 | 42 | self.upc |
701 | 42 | .register_caller( |
702 | 42 | &id, |
703 | 42 | Endpoint { |
704 | 42 | instance: instance.clone(), |
705 | 42 | }, |
706 | 42 | ) |
707 | 42 | .await?0 ; |
708 | 42 | if let Err(err0 ) = self |
709 | 42 | .upc |
710 | 42 | .register_callee( |
711 | 42 | &id, |
712 | 42 | Endpoint { |
713 | 42 | instance: instance.clone(), |
714 | 42 | }, |
715 | | ) |
716 | 42 | .await |
717 | | { |
718 | 0 | self.upc.deregister_caller(&id); |
719 | 0 | return Err(err.into()); |
720 | 42 | } |
721 | 42 | Ok(instance) |
722 | 42 | } |
723 | | |
724 | | /// Deregister a hosted instance from UPC caller and callee tables. |
725 | 6 | pub fn remove_instance(&mut self, id: &LyquidID) { |
726 | 6 | self.upc.deregister_caller(id); |
727 | 6 | self.upc.deregister_callee(id); |
728 | 6 | } |
729 | | } |