urma_node/daemon/
serve.rs

1use super::{
2    ServeOptions,
3    inputs::{invalid, private_directory, read_archive_config, read_header_bootstrap, read_token},
4    overlay::{attach_overlay, shutdown_overlay},
5};
6use crate::{
7    Error,
8    engine::Engine,
9    model::Retention,
10    peer::OverlayCommitSource,
11    service::{Service, ShutdownHandle},
12    source::{CommitSources, LitecoinSource, LocalRpcCommitSource, PublicCommitSource},
13};
14use std::{io::Write, path::PathBuf, sync::Arc};
15use urma_chain::observation::Chain;
16use urma_runtime::{endpoints, p2p::Anchor};
17
18/// Runs the archive service in the foreground until it is asked to stop.
19/// `on_bound` receives the stop handle once the control listener is bound,
20/// before overlay startup and any chain sync, and installs whatever stop
21/// signal the host has (process signals, or the Windows service manager).
22pub fn serve(
23    options: ServeOptions,
24    on_bound: impl FnOnce(ShutdownHandle) -> Result<(), Error>,
25    log: &mut impl Write,
26) -> Result<(), Error> {
27    let config = read_archive_config(&options.config)?;
28    let chain = config.chain;
29    let application_height = config.application_start.height;
30    let mut engine = Engine::open(&options.data, config, None)?;
31    if options.p2p {
32        let source = open_chain_source(&options, chain, application_height)?;
33        engine = engine.with_source(Arc::new(source))?;
34    }
35    let engine = Arc::new(engine);
36    let token = read_token(&options.token_file)?;
37    let retain = options.p2p && engine.status()?.retention == Retention::Unpruned;
38    let mut service = Service::bind(options.listen, token, Arc::clone(&engine))?
39        .with_cache_limit(options.cache_bytes)?;
40    on_bound(service.shutdown_handle())?;
41    if retain {
42        let scratch = options.data.join("scratch");
43        private_directory(&scratch)?;
44        service = service.with_retention(scratch, options.object_timeout)?;
45    }
46    let overlay = (options.overlay, options.p2p);
47    let attached = attach_overlay(service, &options.data, &engine, chain, overlay, log)?;
48    let public = (options.p2p && options.commit_public).then(|| options.data.join("providers"));
49    let local = match (options.p2p, options.commit_rpc) {
50        (true, Some(rpc)) => Some(LocalRpcCommitSource::new(chain, rpc.url, rpc.cookie)),
51        (true, None) | (false, None) => None,
52        (false, Some(rpc)) => {
53            return Err(Error::Invalid(format!(
54                "commit RPC {} needs chain sync",
55                rpc.url
56            )));
57        }
58    };
59    let sources = (local, attached.commit_source, public);
60    let service = attach_commit_sources(attached.service, chain, sources)?;
61    writeln!(
62        log,
63        "URMA Node listening at {} (authenticated loopback)",
64        service.address()?
65    )?;
66    let outcome = service.run(options.p2p);
67    let network_outcome = match attached.network {
68        Some(network) => shutdown_overlay(network),
69        None => Ok(()),
70    };
71    outcome?;
72    network_outcome
73}
74
75fn open_chain_source(
76    options: &ServeOptions,
77    chain: Chain,
78    height: u64,
79) -> Result<LitecoinSource, Error> {
80    let pinned = options.chain_peers.clone();
81    let bootstrap = match &options.header_bootstrap {
82        Some(path) => Some(read_header_bootstrap(path, chain, height)?),
83        None => None,
84    };
85    let cache = options.data.join("headers");
86    std::fs::create_dir_all(&cache)?;
87    match bootstrap {
88        Some(bootstrap) => LitecoinSource::open_bootstrap(&cache, bootstrap, pinned),
89        None => LitecoinSource::open(
90            chain,
91            &cache,
92            Anchor::embedded(chain).map_err(invalid)?,
93            pinned,
94        ),
95    }
96}
97
98type Sources = (
99    Option<LocalRpcCommitSource>,
100    Option<OverlayCommitSource>,
101    Option<PathBuf>,
102);
103
104fn attach_commit_sources(
105    service: Service,
106    chain: Chain,
107    (local, overlay, public): Sources,
108) -> Result<Service, Error> {
109    let mut sources = CommitSources::new(chain);
110    if let Some(local) = local {
111        sources.push("local-rpc".into(), Box::new(local));
112    }
113    if let Some(overlay) = overlay {
114        sources.push("overlay".into(), Box::new(overlay));
115    }
116    let has_public = !endpoints::defaults(chain).is_empty();
117    if let Some(cache) = public.filter(|_cache| has_public) {
118        private_directory(&cache)?;
119        sources.push(
120            "public".into(),
121            Box::new(PublicCommitSource::new(chain, cache)),
122        );
123    }
124    if sources.is_empty() {
125        return Ok(service);
126    }
127    service.with_commit_sources(sources)
128}
urma_node/daemon/serve.rs

urma_node/daemon/
serve.rs
