nornir 0.4.52

Companion to cargo: dependency tracking, release gating, deploy, benchmarks, and documentation assembly. Project-agnostic.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91

//! Embed holger IN-PROCESS — the nornir⇆holger marriage.
//!
//! Instead of spawning `holger-server dev-pair` as a child process, nornir links
//! `holger-server-lib` and runs the **dev pair** (`/cache` crates.io-mirror +
//! writable `/sparring` registry) on a tokio runtime in-process. Three consumers
//! use this: `nornir cache up`, `release stage --execute`, and (optionally) the
//! `nornir-server` daemon at boot. Gated behind the `embed-holger` feature.
//!
//! Holger's own CLI composes startup as:
//!   write `dev_pair_ron` → `read_ron_config` → `instantiate_backends`
//!   → `wire_holger` → `start()`  (then block while the spawned servers run).
//! `start()` is sync and uses `tokio::spawn`, so it must run inside a tokio
//! runtime context; afterwards the gRPC + HTTP listeners live on that runtime.

use anyhow::{Context, Result};
use std::path::Path;
use std::time::Duration;

/// Build the dev-pair config under `data_dir` and start holger's gRPC + HTTP
/// servers on the **current** tokio runtime (the caller must already be inside
/// one — e.g. `nornir-server`'s `#[tokio::main]`). Returns as soon as the servers
/// are spawned; use [`wait_ready`] to confirm the HTTP gateway is accepting.
pub fn start_in_current_runtime(data_dir: &Path, grpc: &str, http: &str) -> Result<()> {
    std::fs::create_dir_all(data_dir)
        .with_context(|| format!("create holger data dir {}", data_dir.display()))?;
    let cfg_path = data_dir.join("holger-dev-pair.ron");
    std::fs::write(&cfg_path, holger_server_lib::dev_pair_ron(data_dir, grpc, http))
        .with_context(|| format!("write holger config {}", cfg_path.display()))?;
    let mut holger = holger_server_lib::read_ron_config(&cfg_path)
        .with_context(|| format!("read holger config {}", cfg_path.display()))?;
    holger.instantiate_backends().context("holger: instantiate backends")?;
    holger_server_lib::wire_holger(&mut holger).context("holger: wire routes")?;
    holger.start().context("holger: start servers")?;
    Ok(())
}

/// A holger dev-pair running on its **own** tokio runtime — for the sync CLI
/// paths (`nornir cache up`, `release stage`) that aren't already inside a
/// runtime. Hold this for as long as the servers should run; dropping it drops
/// the runtime and stops the servers (the rehearsal registry is ephemeral).
pub struct EmbeddedHolger {
    /// Owns the worker threads the gRPC/HTTP listeners run on. Dropping it (last
    /// field) shuts them down.
    _rt: tokio::runtime::Runtime,
    pub grpc: String,
    pub http: String,
}

impl EmbeddedHolger {
    /// Spin up a dedicated multi-thread runtime, start the dev pair on it, and
    /// return the live handle. Does NOT wait for readiness — call
    /// [`EmbeddedHolger::wait_ready`].
    pub fn start(data_dir: &Path, grpc: &str, http: &str) -> Result<Self> {
        let rt = tokio::runtime::Builder::new_multi_thread()
            .enable_all()
            .worker_threads(2)
            .thread_name("holger-embed")
            .build()
            .context("build holger tokio runtime")?;
        {
            let _guard = rt.enter();
            start_in_current_runtime(data_dir, grpc, http)?;
        }
        Ok(Self { _rt: rt, grpc: grpc.to_string(), http: http.to_string() })
    }

    /// Poll the HTTP gateway until the sparse index serves (or time out).
    pub fn wait_ready(&self, timeout: Duration) -> Result<()> {
        wait_ready(&self.http, timeout)
    }
}

/// Poll holger's HTTP sparse gateway at `http_addr` (host:port) until it serves
/// the `/cache` index `config.json` (proof the routes are wired), or time out.
pub fn wait_ready(http_addr: &str, timeout: Duration) -> Result<()> {
    let url = format!("http://{http_addr}/cache/index/config.json");
    let deadline = std::time::Instant::now() + timeout;
    loop {
        if ureq::get(&url)
            .timeout(Duration::from_millis(500))
            .call()
            .is_ok()
        {
            return Ok(());
        }
        if std::time::Instant::now() >= deadline {
            anyhow::bail!("holger HTTP gateway never became ready at {url} (waited {timeout:?})");
        }
        std::thread::sleep(Duration::from_millis(150));
    }
}