Coverage Report

Created: 2026-07-26 02:19

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
/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
}