Archive Rust bootstrap — El compiler is now self-hosting
This commit is contained in:
Will Anderson
@@ -0,0 +1,48 @@
|
||||
[package]
|
||||
name = "el"
|
||||
description = "Engram language CLI — el build / run / check / seal / unseal / new / add / publish / …"
|
||||
version.workspace = true
|
||||
edition.workspace = true
|
||||
license.workspace = true
|
||||
|
||||
[[bin]]
|
||||
name = "el"
|
||||
path = "src/main.rs"
|
||||
|
||||
[dependencies]
|
||||
el-lexer = { workspace = true }
|
||||
el-parser = { workspace = true }
|
||||
el-types = { workspace = true }
|
||||
el-compiler = { workspace = true }
|
||||
el-seal = { workspace = true }
|
||||
el-manifest = { workspace = true }
|
||||
el-registry = { workspace = true }
|
||||
el-build = { workspace = true }
|
||||
el-test = { workspace = true }
|
||||
el-fmt = { workspace = true }
|
||||
el-lint = { workspace = true }
|
||||
clap = { workspace = true }
|
||||
thiserror = { workspace = true }
|
||||
serde_json = { workspace = true }
|
||||
serde = { workspace = true }
|
||||
tokio = { version = "1", features = ["rt", "rt-multi-thread", "macros"] }
|
||||
reqwest = { workspace = true }
|
||||
uuid = { workspace = true }
|
||||
walkdir = { workspace = true }
|
||||
tiny_http = { workspace = true }
|
||||
blake3 = { workspace = true }
|
||||
hmac = { workspace = true }
|
||||
sha2 = { workspace = true }
|
||||
hex = { workspace = true }
|
||||
base64 = { workspace = true }
|
||||
aes-gcm = { workspace = true }
|
||||
rand = { workspace = true }
|
||||
ed25519-dalek = { workspace = true }
|
||||
crossbeam-channel = { workspace = true }
|
||||
winit = { version = "0.29", default-features = false, features = ["rwh_05"] }
|
||||
softbuffer = "0.3"
|
||||
tiny-skia = "0.11"
|
||||
fontdue = "0.8"
|
||||
image = { workspace = true }
|
||||
tungstenite = { workspace = true }
|
||||
native-tls = { workspace = true }
|
||||
@@ -0,0 +1,130 @@
|
||||
//! In-memory LRU cache for HTTP responses and LLM outputs.
|
||||
//!
|
||||
//! Thread-safe global cache keyed by arbitrary strings. Entries expire after a
|
||||
//! configurable TTL. Eviction uses a simple LRU strategy: when the entry limit
|
||||
//! is reached the oldest entry (by insertion / last-access order) is dropped.
|
||||
//!
|
||||
//! This module only contains the cache data-structure and its helper
|
||||
//! functions. The builtin dispatch arms that expose `cache_get`, `cache_set`,
|
||||
//! `cache_invalidate`, and `cache_clear` live in `main.rs`.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::sync::{Mutex, OnceLock};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
const MAX_ENTRIES: usize = 1000;
|
||||
|
||||
/// A single cached value together with its expiry time.
|
||||
struct Entry {
|
||||
value: String,
|
||||
expires_at: Instant,
|
||||
/// Monotonically increasing sequence number used to identify the LRU entry.
|
||||
seq: u64,
|
||||
}
|
||||
|
||||
struct Cache {
|
||||
entries: HashMap<String, Entry>,
|
||||
/// Counter incremented on every write; stored in Entry::seq.
|
||||
seq: u64,
|
||||
}
|
||||
|
||||
impl Cache {
|
||||
fn new() -> Self {
|
||||
Cache {
|
||||
entries: HashMap::new(),
|
||||
seq: 0,
|
||||
}
|
||||
}
|
||||
|
||||
/// Store `value` under `key`, expiring after `ttl_secs` seconds.
|
||||
/// If the cache is full the LRU entry is evicted before insertion.
|
||||
pub fn set(&mut self, key: String, value: String, ttl_secs: u64) {
|
||||
// Evict expired entries first (cheap pass).
|
||||
let now = Instant::now();
|
||||
self.entries.retain(|_, e| e.expires_at > now);
|
||||
|
||||
// If still full, evict the entry with the lowest sequence number (LRU).
|
||||
if self.entries.len() >= MAX_ENTRIES {
|
||||
if let Some(lru_key) = self
|
||||
.entries
|
||||
.iter()
|
||||
.min_by_key(|(_, e)| e.seq)
|
||||
.map(|(k, _)| k.clone())
|
||||
{
|
||||
self.entries.remove(&lru_key);
|
||||
}
|
||||
}
|
||||
|
||||
self.seq += 1;
|
||||
self.entries.insert(
|
||||
key,
|
||||
Entry {
|
||||
value,
|
||||
expires_at: now + Duration::from_secs(ttl_secs),
|
||||
seq: self.seq,
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
/// Return the cached value if it exists and has not expired, otherwise `None`.
|
||||
pub fn get(&mut self, key: &str) -> Option<String> {
|
||||
let now = Instant::now();
|
||||
if let Some(e) = self.entries.get_mut(key) {
|
||||
if e.expires_at > now {
|
||||
// Bump sequence number to record this access (LRU).
|
||||
self.seq += 1;
|
||||
e.seq = self.seq;
|
||||
return Some(e.value.clone());
|
||||
}
|
||||
// Expired — remove it.
|
||||
self.entries.remove(key);
|
||||
}
|
||||
None
|
||||
}
|
||||
|
||||
/// Remove a specific key.
|
||||
pub fn invalidate(&mut self, key: &str) {
|
||||
self.entries.remove(key);
|
||||
}
|
||||
|
||||
/// Remove all entries.
|
||||
pub fn clear(&mut self) {
|
||||
self.entries.clear();
|
||||
}
|
||||
}
|
||||
|
||||
// ── Global singleton ──────────────────────────────────────────────────────────
|
||||
|
||||
static CACHE: OnceLock<Mutex<Cache>> = OnceLock::new();
|
||||
|
||||
fn global() -> &'static Mutex<Cache> {
|
||||
CACHE.get_or_init(|| Mutex::new(Cache::new()))
|
||||
}
|
||||
|
||||
// ── Public API ────────────────────────────────────────────────────────────────
|
||||
|
||||
/// Store an arbitrary string value under `key` with a TTL in seconds.
|
||||
pub fn cache_set(key: String, value: String, ttl_secs: u64) {
|
||||
if let Ok(mut c) = global().lock() {
|
||||
c.set(key, value, ttl_secs);
|
||||
}
|
||||
}
|
||||
|
||||
/// Return the cached value for `key`, or `None` if missing / expired.
|
||||
pub fn cache_get(key: &str) -> Option<String> {
|
||||
global().lock().ok()?.get(key)
|
||||
}
|
||||
|
||||
/// Remove a specific key from the cache.
|
||||
pub fn cache_invalidate(key: &str) {
|
||||
if let Ok(mut c) = global().lock() {
|
||||
c.invalidate(key);
|
||||
}
|
||||
}
|
||||
|
||||
/// Remove all entries from the cache.
|
||||
pub fn cache_clear() {
|
||||
if let Ok(mut c) = global().lock() {
|
||||
c.clear();
|
||||
}
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,730 @@
|
||||
//! Networking enhancements: HTTP retry/backoff, circuit breaker, and WebSocket server.
|
||||
//!
|
||||
//! Everything here is synchronous / blocking to match the rest of the El runtime
|
||||
//! (which uses `reqwest::blocking` throughout). WebSocket server connections run
|
||||
//! on background OS threads; the interpreter thread itself never blocks waiting
|
||||
//! for the server.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::net::TcpListener;
|
||||
use std::sync::{Arc, Mutex, OnceLock};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// HTTP retry with exponential back-off
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
/// Perform an HTTP GET, retrying on 5xx responses or connection errors.
|
||||
///
|
||||
/// `max_attempts` — total number of attempts (1 = no retry).
|
||||
/// `backoff_ms` — initial back-off in milliseconds; doubles each attempt.
|
||||
///
|
||||
/// Returns the response body on the first successful (non-5xx) response, or an
|
||||
/// error JSON string after all attempts are exhausted.
|
||||
pub fn http_get_retry(url: &str, max_attempts: u32, backoff_ms: u64) -> String {
|
||||
let client = reqwest::blocking::Client::builder()
|
||||
.timeout(Duration::from_secs(30))
|
||||
.build()
|
||||
.unwrap_or_default();
|
||||
|
||||
let mut delay = backoff_ms;
|
||||
for attempt in 0..max_attempts.max(1) {
|
||||
match client.get(url).send() {
|
||||
Ok(resp) => {
|
||||
let status = resp.status().as_u16();
|
||||
let body = resp.text().unwrap_or_default();
|
||||
if status < 500 {
|
||||
return body;
|
||||
}
|
||||
// 5xx — retry unless this was the last attempt
|
||||
if attempt + 1 < max_attempts {
|
||||
std::thread::sleep(Duration::from_millis(delay));
|
||||
delay *= 2;
|
||||
} else {
|
||||
return format!(
|
||||
"{{\"error\":\"http_get_retry: server error {status} after {max_attempts} attempt(s)\",\"body\":{body:?}}}"
|
||||
);
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
if attempt + 1 < max_attempts {
|
||||
std::thread::sleep(Duration::from_millis(delay));
|
||||
delay *= 2;
|
||||
} else {
|
||||
return format!(
|
||||
"{{\"error\":\"http_get_retry: {e} after {max_attempts} attempt(s)\"}}"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
format!("{{\"error\":\"http_get_retry: no attempts executed\"}}")
|
||||
}
|
||||
|
||||
/// Perform an HTTP POST with JSON body, retrying on 5xx or connection errors.
|
||||
///
|
||||
/// Same retry semantics as [`http_get_retry`].
|
||||
pub fn http_post_retry(url: &str, body: &str, max_attempts: u32, backoff_ms: u64) -> String {
|
||||
let client = reqwest::blocking::Client::builder()
|
||||
.timeout(Duration::from_secs(30))
|
||||
.build()
|
||||
.unwrap_or_default();
|
||||
|
||||
let mut delay = backoff_ms;
|
||||
for attempt in 0..max_attempts.max(1) {
|
||||
match client
|
||||
.post(url)
|
||||
.header("Content-Type", "application/json")
|
||||
.body(body.to_owned())
|
||||
.send()
|
||||
{
|
||||
Ok(resp) => {
|
||||
let status = resp.status().as_u16();
|
||||
let resp_body = resp.text().unwrap_or_default();
|
||||
if status < 500 {
|
||||
return resp_body;
|
||||
}
|
||||
if attempt + 1 < max_attempts {
|
||||
std::thread::sleep(Duration::from_millis(delay));
|
||||
delay *= 2;
|
||||
} else {
|
||||
return format!(
|
||||
"{{\"error\":\"http_post_retry: server error {status} after {max_attempts} attempt(s)\",\"body\":{resp_body:?}}}"
|
||||
);
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
if attempt + 1 < max_attempts {
|
||||
std::thread::sleep(Duration::from_millis(delay));
|
||||
delay *= 2;
|
||||
} else {
|
||||
return format!(
|
||||
"{{\"error\":\"http_post_retry: {e} after {max_attempts} attempt(s)\"}}"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
format!("{{\"error\":\"http_post_retry: no attempts executed\"}}")
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Circuit breaker
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
#[derive(Debug, Clone, PartialEq)]
|
||||
enum CircuitState {
|
||||
/// Passing requests through normally.
|
||||
Closed,
|
||||
/// Too many failures — reject immediately.
|
||||
Open {
|
||||
/// When the circuit may attempt to close again.
|
||||
until: Instant,
|
||||
},
|
||||
/// One probe request allowed through; waiting to see if it succeeds.
|
||||
HalfOpen,
|
||||
}
|
||||
|
||||
struct CircuitBreaker {
|
||||
state: CircuitState,
|
||||
failure_count: u32,
|
||||
failure_threshold: u32,
|
||||
reset_duration: Duration,
|
||||
}
|
||||
|
||||
impl CircuitBreaker {
|
||||
fn new(failure_threshold: u32, reset_secs: u64) -> Self {
|
||||
CircuitBreaker {
|
||||
state: CircuitState::Closed,
|
||||
failure_count: 0,
|
||||
failure_threshold,
|
||||
reset_duration: Duration::from_secs(reset_secs),
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns `true` if the circuit should allow a call through right now.
|
||||
fn allow(&mut self) -> bool {
|
||||
match &self.state {
|
||||
CircuitState::Closed => true,
|
||||
CircuitState::Open { until } => {
|
||||
if Instant::now() >= *until {
|
||||
self.state = CircuitState::HalfOpen;
|
||||
true
|
||||
} else {
|
||||
false
|
||||
}
|
||||
}
|
||||
CircuitState::HalfOpen => true,
|
||||
}
|
||||
}
|
||||
|
||||
fn record_success(&mut self) {
|
||||
self.failure_count = 0;
|
||||
self.state = CircuitState::Closed;
|
||||
}
|
||||
|
||||
fn record_failure(&mut self) {
|
||||
self.failure_count += 1;
|
||||
if self.failure_count >= self.failure_threshold
|
||||
|| self.state == CircuitState::HalfOpen
|
||||
{
|
||||
self.state = CircuitState::Open {
|
||||
until: Instant::now() + self.reset_duration,
|
||||
};
|
||||
self.failure_count = 0;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ── Global circuit-breaker registry ──────────────────────────────────────────
|
||||
|
||||
static CIRCUITS: OnceLock<Mutex<HashMap<String, CircuitBreaker>>> = OnceLock::new();
|
||||
|
||||
fn circuits() -> &'static Mutex<HashMap<String, CircuitBreaker>> {
|
||||
CIRCUITS.get_or_init(|| Mutex::new(HashMap::new()))
|
||||
}
|
||||
|
||||
/// Register (or replace) a circuit breaker with the given name.
|
||||
pub fn circuit_open(name: String, failure_threshold: u32, reset_secs: u64) {
|
||||
if let Ok(mut map) = circuits().lock() {
|
||||
map.insert(name, CircuitBreaker::new(failure_threshold, reset_secs));
|
||||
}
|
||||
}
|
||||
|
||||
/// Make an HTTP POST call through the named circuit breaker.
|
||||
///
|
||||
/// * If the circuit is open: returns `{"error":"circuit open"}` immediately.
|
||||
/// * If closed/half-open: makes the POST, records success or failure, and
|
||||
/// returns the response body.
|
||||
pub fn circuit_call(name: &str, url: &str, body: &str) -> String {
|
||||
// Check + allow atomically under the lock, then drop the lock before the
|
||||
// blocking network call (which could take seconds).
|
||||
let allowed = circuits()
|
||||
.lock()
|
||||
.ok()
|
||||
.and_then(|mut map| map.get_mut(name).map(|cb| cb.allow()))
|
||||
.unwrap_or(true); // unknown circuit name → allow
|
||||
|
||||
if !allowed {
|
||||
return r#"{"error":"circuit open"}"#.to_owned();
|
||||
}
|
||||
|
||||
let client = reqwest::blocking::Client::builder()
|
||||
.timeout(Duration::from_secs(30))
|
||||
.build()
|
||||
.unwrap_or_default();
|
||||
|
||||
match client
|
||||
.post(url)
|
||||
.header("Content-Type", "application/json")
|
||||
.body(body.to_owned())
|
||||
.send()
|
||||
{
|
||||
Ok(resp) => {
|
||||
let status = resp.status().as_u16();
|
||||
let resp_body = resp.text().unwrap_or_default();
|
||||
if status >= 500 {
|
||||
if let Ok(mut map) = circuits().lock() {
|
||||
if let Some(cb) = map.get_mut(name) {
|
||||
cb.record_failure();
|
||||
}
|
||||
}
|
||||
format!("{{\"error\":\"circuit_call: server error {status}\",\"body\":{resp_body:?}}}")
|
||||
} else {
|
||||
if let Ok(mut map) = circuits().lock() {
|
||||
if let Some(cb) = map.get_mut(name) {
|
||||
cb.record_success();
|
||||
}
|
||||
}
|
||||
resp_body
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
if let Ok(mut map) = circuits().lock() {
|
||||
if let Some(cb) = map.get_mut(name) {
|
||||
cb.record_failure();
|
||||
}
|
||||
}
|
||||
format!("{{\"error\":\"circuit_call: {e}\"}}")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// WebSocket server
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
//
|
||||
// Implementation strategy
|
||||
// ───────────────────────
|
||||
// The El interpreter is single-threaded and synchronous. We need a WebSocket
|
||||
// *server* that can accept multiple concurrent clients while the interpreter
|
||||
// keeps running.
|
||||
//
|
||||
// Solution: each accepted connection runs on its own OS thread. Outbound
|
||||
// messages are queued through a per-client `mpsc` channel. The interpreter
|
||||
// calls `ws_serve`, `ws_send`, `ws_broadcast`, and `ws_close` — all of which
|
||||
// return immediately and coordinate with the background threads via shared
|
||||
// state.
|
||||
//
|
||||
// Handler callbacks are *not* invoked on the background threads; instead,
|
||||
// incoming messages are placed in a global queue and the interpreter drains
|
||||
// them by calling `ws_poll` (or they are dispatched automatically inside a
|
||||
// future tight-loop variant of `ws_serve`).
|
||||
//
|
||||
// For simplicity this implementation uses `tungstenite` (synchronous), the
|
||||
// same crate the existing `ws_connect` client already uses.
|
||||
|
||||
use std::sync::mpsc;
|
||||
|
||||
/// Message queued from a background connection thread to the interpreter.
|
||||
#[derive(Debug)]
|
||||
pub struct IncomingWsMessage {
|
||||
pub client_id: String,
|
||||
pub message: String,
|
||||
}
|
||||
|
||||
/// Represents one connected WebSocket client.
|
||||
struct WsClient {
|
||||
/// Channel sender used to push outbound messages to the background thread.
|
||||
tx: mpsc::Sender<Option<String>>, // None = disconnect signal
|
||||
}
|
||||
|
||||
/// Shared server state accessible from both the interpreter thread and the
|
||||
/// background connection threads.
|
||||
pub struct WsServerState {
|
||||
/// Connected clients, keyed by client_id.
|
||||
clients: HashMap<String, WsClient>,
|
||||
/// Messages received from clients waiting to be delivered to the handler.
|
||||
inbox: Vec<IncomingWsMessage>,
|
||||
/// Counter for generating unique client IDs.
|
||||
next_id: u64,
|
||||
}
|
||||
|
||||
impl WsServerState {
|
||||
fn new() -> Self {
|
||||
WsServerState {
|
||||
clients: HashMap::new(),
|
||||
inbox: Vec::new(),
|
||||
next_id: 1,
|
||||
}
|
||||
}
|
||||
|
||||
fn next_client_id(&mut self) -> String {
|
||||
let id = format!("wsc:{}", self.next_id);
|
||||
self.next_id += 1;
|
||||
id
|
||||
}
|
||||
}
|
||||
|
||||
// ── Global server state registry (one entry per listening port) ───────────────
|
||||
|
||||
static WS_SERVERS: OnceLock<Mutex<HashMap<u16, Arc<Mutex<WsServerState>>>>> = OnceLock::new();
|
||||
|
||||
fn ws_servers() -> &'static Mutex<HashMap<u16, Arc<Mutex<WsServerState>>>> {
|
||||
WS_SERVERS.get_or_init(|| Mutex::new(HashMap::new()))
|
||||
}
|
||||
|
||||
fn get_or_create_server(port: u16) -> Arc<Mutex<WsServerState>> {
|
||||
let mut map = ws_servers().lock().unwrap();
|
||||
map.entry(port)
|
||||
.or_insert_with(|| Arc::new(Mutex::new(WsServerState::new())))
|
||||
.clone()
|
||||
}
|
||||
|
||||
// ── Public API (called from dispatch_builtin) ─────────────────────────────────
|
||||
|
||||
/// Start a WebSocket server on `port`.
|
||||
///
|
||||
/// This function spawns a background acceptor thread and returns immediately.
|
||||
/// Incoming connections are handled on per-connection threads. Messages
|
||||
/// received are pushed into the server's inbox; call [`ws_poll`] to drain them.
|
||||
pub fn ws_serve_start(port: u16) {
|
||||
let state = get_or_create_server(port);
|
||||
|
||||
// Spawn acceptor thread.
|
||||
let state_clone = state.clone();
|
||||
std::thread::spawn(move || {
|
||||
let listener = match TcpListener::bind(format!("0.0.0.0:{port}")) {
|
||||
Ok(l) => l,
|
||||
Err(e) => {
|
||||
eprintln!("[ws_serve] failed to bind port {port}: {e}");
|
||||
return;
|
||||
}
|
||||
};
|
||||
eprintln!("[ws_serve] listening on port {port}");
|
||||
|
||||
for stream in listener.incoming() {
|
||||
match stream {
|
||||
Ok(tcp) => {
|
||||
let state_for_conn = state_clone.clone();
|
||||
std::thread::spawn(move || {
|
||||
handle_ws_connection(tcp, state_for_conn);
|
||||
});
|
||||
}
|
||||
Err(e) => {
|
||||
eprintln!("[ws_serve] accept error: {e}");
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
fn handle_ws_connection(stream: std::net::TcpStream, state: Arc<Mutex<WsServerState>>) {
|
||||
let _ = stream.set_read_timeout(Some(Duration::from_millis(100)));
|
||||
|
||||
let ws_result = tungstenite::accept(stream);
|
||||
let mut ws = match ws_result {
|
||||
Ok(w) => w,
|
||||
Err(e) => {
|
||||
eprintln!("[ws_serve] WebSocket handshake error: {e}");
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
// Assign a client ID.
|
||||
let client_id = {
|
||||
let mut s = state.lock().unwrap();
|
||||
let id = s.next_client_id();
|
||||
// We'll register the sender after we create the channel below.
|
||||
id
|
||||
};
|
||||
|
||||
// Create an outbound channel for this connection.
|
||||
let (tx, rx) = mpsc::channel::<Option<String>>();
|
||||
|
||||
{
|
||||
let mut s = state.lock().unwrap();
|
||||
s.clients.insert(client_id.clone(), WsClient { tx });
|
||||
}
|
||||
|
||||
eprintln!("[ws_serve] client connected: {client_id}");
|
||||
|
||||
loop {
|
||||
// --- Receive incoming messages (non-blocking with short timeout) ---
|
||||
match ws.read() {
|
||||
Ok(tungstenite::Message::Text(text)) => {
|
||||
let mut s = state.lock().unwrap();
|
||||
s.inbox.push(IncomingWsMessage {
|
||||
client_id: client_id.clone(),
|
||||
message: text.to_string(),
|
||||
});
|
||||
}
|
||||
Ok(tungstenite::Message::Binary(bytes)) => {
|
||||
let text = String::from_utf8_lossy(&bytes).into_owned();
|
||||
let mut s = state.lock().unwrap();
|
||||
s.inbox.push(IncomingWsMessage {
|
||||
client_id: client_id.clone(),
|
||||
message: text,
|
||||
});
|
||||
}
|
||||
Ok(tungstenite::Message::Close(_)) => {
|
||||
break;
|
||||
}
|
||||
Ok(tungstenite::Message::Ping(data)) => {
|
||||
let _ = ws.send(tungstenite::Message::Pong(data));
|
||||
}
|
||||
Ok(_) => {}
|
||||
Err(tungstenite::Error::Io(e))
|
||||
if e.kind() == std::io::ErrorKind::WouldBlock
|
||||
|| e.kind() == std::io::ErrorKind::TimedOut =>
|
||||
{
|
||||
// No data yet — check the outbound channel.
|
||||
}
|
||||
Err(e) => {
|
||||
eprintln!("[ws_serve] read error for {client_id}: {e}");
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
// --- Send queued outbound messages ---
|
||||
loop {
|
||||
match rx.try_recv() {
|
||||
Ok(Some(msg)) => {
|
||||
if ws
|
||||
.send(tungstenite::Message::Text(msg.into()))
|
||||
.is_err()
|
||||
{
|
||||
break;
|
||||
}
|
||||
}
|
||||
Ok(None) => {
|
||||
// Disconnect signal.
|
||||
let _ = ws.close(None);
|
||||
let mut s = state.lock().unwrap();
|
||||
s.clients.remove(&client_id);
|
||||
eprintln!("[ws_serve] client disconnected (server-side): {client_id}");
|
||||
return;
|
||||
}
|
||||
Err(mpsc::TryRecvError::Empty) => break,
|
||||
Err(mpsc::TryRecvError::Disconnected) => {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Clean up.
|
||||
{
|
||||
let mut s = state.lock().unwrap();
|
||||
s.clients.remove(&client_id);
|
||||
}
|
||||
eprintln!("[ws_serve] client disconnected: {client_id}");
|
||||
}
|
||||
|
||||
/// Send a message to a specific connected client.
|
||||
/// Returns `false` if the client is not found.
|
||||
pub fn ws_server_send(port: u16, client_id: &str, message: String) -> bool {
|
||||
let map = ws_servers().lock().unwrap();
|
||||
if let Some(state) = map.get(&port) {
|
||||
let s = state.lock().unwrap();
|
||||
if let Some(client) = s.clients.get(client_id) {
|
||||
return client.tx.send(Some(message)).is_ok();
|
||||
}
|
||||
}
|
||||
false
|
||||
}
|
||||
|
||||
/// Broadcast a message to all connected clients on a port.
|
||||
pub fn ws_server_broadcast(port: u16, message: String) {
|
||||
let map = ws_servers().lock().unwrap();
|
||||
if let Some(state) = map.get(&port) {
|
||||
let s = state.lock().unwrap();
|
||||
for client in s.clients.values() {
|
||||
let _ = client.tx.send(Some(message.clone()));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Disconnect a specific client.
|
||||
pub fn ws_server_close(port: u16, client_id: &str) {
|
||||
let map = ws_servers().lock().unwrap();
|
||||
if let Some(state) = map.get(&port) {
|
||||
let s = state.lock().unwrap();
|
||||
if let Some(client) = s.clients.get(client_id) {
|
||||
let _ = client.tx.send(None);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Drain pending incoming messages for a given server port.
|
||||
///
|
||||
/// Returns up to `max` messages (or all if `max == 0`). The caller is
|
||||
/// responsible for invoking the El handler function for each message.
|
||||
pub fn ws_server_poll(port: u16, max: usize) -> Vec<IncomingWsMessage> {
|
||||
let map = ws_servers().lock().unwrap();
|
||||
if let Some(state) = map.get(&port) {
|
||||
let mut s = state.lock().unwrap();
|
||||
if max == 0 || s.inbox.len() <= max {
|
||||
let msgs = std::mem::take(&mut s.inbox);
|
||||
msgs
|
||||
} else {
|
||||
s.inbox.drain(..max).collect()
|
||||
}
|
||||
} else {
|
||||
vec![]
|
||||
}
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// WebSocket client with handler callback support
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
//
|
||||
// `ws_connect` already exists in `main.rs` using synchronous tungstenite.
|
||||
// This new variant (`ws_connect_handler`) runs the connection on a background
|
||||
// thread and queues incoming messages so `ws_client_poll` can deliver them to
|
||||
// the El handler.
|
||||
|
||||
/// Pending message from a background WS client connection.
|
||||
#[derive(Debug)]
|
||||
pub struct IncomingClientMessage {
|
||||
pub conn_id: String,
|
||||
pub message: String,
|
||||
}
|
||||
|
||||
struct WsClientConn {
|
||||
tx: mpsc::Sender<Option<String>>,
|
||||
}
|
||||
|
||||
static WS_CLIENT_CONNS: OnceLock<Mutex<HashMap<String, WsClientConn>>> = OnceLock::new();
|
||||
static WS_CLIENT_INBOX: OnceLock<Mutex<Vec<IncomingClientMessage>>> = OnceLock::new();
|
||||
static WS_CLIENT_COUNTER: OnceLock<Mutex<u64>> = OnceLock::new();
|
||||
|
||||
fn ws_client_conns() -> &'static Mutex<HashMap<String, WsClientConn>> {
|
||||
WS_CLIENT_CONNS.get_or_init(|| Mutex::new(HashMap::new()))
|
||||
}
|
||||
|
||||
fn ws_client_inbox() -> &'static Mutex<Vec<IncomingClientMessage>> {
|
||||
WS_CLIENT_INBOX.get_or_init(|| Mutex::new(Vec::new()))
|
||||
}
|
||||
|
||||
fn next_conn_id() -> String {
|
||||
let mut ctr = WS_CLIENT_COUNTER
|
||||
.get_or_init(|| Mutex::new(1))
|
||||
.lock()
|
||||
.unwrap();
|
||||
let id = format!("wsconn:{}", *ctr);
|
||||
*ctr += 1;
|
||||
id
|
||||
}
|
||||
|
||||
/// Connect to a WebSocket server in the background.
|
||||
///
|
||||
/// Returns a `conn_id` string immediately. Incoming messages are queued and
|
||||
/// can be retrieved with [`ws_client_poll`].
|
||||
pub fn ws_client_connect(url: &str) -> String {
|
||||
let conn_id = next_conn_id();
|
||||
let (tx, rx) = mpsc::channel::<Option<String>>();
|
||||
|
||||
{
|
||||
let mut map = ws_client_conns().lock().unwrap();
|
||||
map.insert(conn_id.clone(), WsClientConn { tx });
|
||||
}
|
||||
|
||||
let url_owned = url.to_owned();
|
||||
let conn_id_clone = conn_id.clone();
|
||||
|
||||
std::thread::spawn(move || {
|
||||
let ws_result = tungstenite::connect(&url_owned);
|
||||
let (mut ws, _) = match ws_result {
|
||||
Ok(pair) => pair,
|
||||
Err(e) => {
|
||||
eprintln!("[ws_client] connect error for {conn_id_clone}: {e}");
|
||||
let mut map = ws_client_conns().lock().unwrap();
|
||||
map.remove(&conn_id_clone);
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
// Set a short read timeout so the loop stays responsive to outbound messages.
|
||||
// MaybeTlsStream exposes the underlying stream via get_ref/get_mut.
|
||||
{
|
||||
use tungstenite::stream::MaybeTlsStream;
|
||||
match ws.get_mut() {
|
||||
MaybeTlsStream::Plain(tcp) => {
|
||||
let _ = tcp.set_read_timeout(Some(Duration::from_millis(50)));
|
||||
}
|
||||
#[cfg(feature = "native-tls")]
|
||||
MaybeTlsStream::NativeTls(tls) => {
|
||||
let _ = tls.get_ref().set_read_timeout(Some(Duration::from_millis(50)));
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
|
||||
loop {
|
||||
// Receive from server.
|
||||
match ws.read() {
|
||||
Ok(tungstenite::Message::Text(text)) => {
|
||||
let mut inbox = ws_client_inbox().lock().unwrap();
|
||||
inbox.push(IncomingClientMessage {
|
||||
conn_id: conn_id_clone.clone(),
|
||||
message: text.to_string(),
|
||||
});
|
||||
}
|
||||
Ok(tungstenite::Message::Binary(bytes)) => {
|
||||
let text = String::from_utf8_lossy(&bytes).into_owned();
|
||||
let mut inbox = ws_client_inbox().lock().unwrap();
|
||||
inbox.push(IncomingClientMessage {
|
||||
conn_id: conn_id_clone.clone(),
|
||||
message: text,
|
||||
});
|
||||
}
|
||||
Ok(tungstenite::Message::Close(_)) => break,
|
||||
Ok(tungstenite::Message::Ping(data)) => {
|
||||
let _ = ws.send(tungstenite::Message::Pong(data));
|
||||
}
|
||||
Ok(_) => {}
|
||||
Err(tungstenite::Error::Io(e))
|
||||
if e.kind() == std::io::ErrorKind::WouldBlock
|
||||
|| e.kind() == std::io::ErrorKind::TimedOut =>
|
||||
{
|
||||
// No data yet.
|
||||
}
|
||||
Err(e) => {
|
||||
eprintln!("[ws_client] read error for {conn_id_clone}: {e}");
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
// Send queued outbound messages.
|
||||
loop {
|
||||
match rx.try_recv() {
|
||||
Ok(Some(msg)) => {
|
||||
if ws.send(tungstenite::Message::Text(msg.into())).is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
Ok(None) => {
|
||||
let _ = ws.close(None);
|
||||
let mut map = ws_client_conns().lock().unwrap();
|
||||
map.remove(&conn_id_clone);
|
||||
return;
|
||||
}
|
||||
Err(mpsc::TryRecvError::Empty) => break,
|
||||
Err(mpsc::TryRecvError::Disconnected) => break,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let mut map = ws_client_conns().lock().unwrap();
|
||||
map.remove(&conn_id_clone);
|
||||
});
|
||||
|
||||
conn_id
|
||||
}
|
||||
|
||||
/// Send a message on an existing client connection.
|
||||
pub fn ws_client_send(conn_id: &str, message: String) {
|
||||
let map = ws_client_conns().lock().unwrap();
|
||||
if let Some(conn) = map.get(conn_id) {
|
||||
let _ = conn.tx.send(Some(message));
|
||||
}
|
||||
}
|
||||
|
||||
/// Close a client connection.
|
||||
pub fn ws_client_close(conn_id: &str) {
|
||||
let map = ws_client_conns().lock().unwrap();
|
||||
if let Some(conn) = map.get(conn_id) {
|
||||
let _ = conn.tx.send(None);
|
||||
}
|
||||
}
|
||||
|
||||
/// Drain pending incoming messages for client connections.
|
||||
///
|
||||
/// Returns all queued messages (all connections combined). Pass `conn_id` as
|
||||
/// `Some(&str)` to filter to a specific connection, or `None` for all.
|
||||
pub fn ws_client_poll(conn_id_filter: Option<&str>) -> Vec<IncomingClientMessage> {
|
||||
let mut inbox = ws_client_inbox().lock().unwrap();
|
||||
if let Some(filter) = conn_id_filter {
|
||||
let (matching, rest): (Vec<_>, Vec<_>) =
|
||||
std::mem::take(&mut *inbox)
|
||||
.into_iter()
|
||||
.partition(|m| m.conn_id == filter);
|
||||
*inbox = rest;
|
||||
matching
|
||||
} else {
|
||||
std::mem::take(&mut *inbox)
|
||||
}
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Transient retry wrapper (for enhancing existing http_get / http_post)
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
/// Try `f` once; on connection error, wait `delay_ms` and try once more.
|
||||
///
|
||||
/// This is the "automatic single retry on transient network error" behaviour
|
||||
/// layered over the existing `http_get` / `http_post` builtins.
|
||||
pub fn with_single_retry<F>(delay_ms: u64, f: F) -> String
|
||||
where
|
||||
F: Fn() -> Result<String, reqwest::Error>,
|
||||
{
|
||||
match f() {
|
||||
Ok(body) => body,
|
||||
Err(e) if e.is_connect() || e.is_timeout() => {
|
||||
std::thread::sleep(Duration::from_millis(delay_ms));
|
||||
f().unwrap_or_else(|e2| format!("{{\"error\":\"{e2}\"}}"))
|
||||
}
|
||||
Err(e) => format!("{{\"error\":\"{e}\"}}"),
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,641 @@
|
||||
//! Automatic, zero-config observability for the El runtime.
|
||||
//!
|
||||
//! This module implements:
|
||||
//! - A background OTLP/HTTP exporter (spans + logs + metrics)
|
||||
//! - Thread-local span context for propagation
|
||||
//! - Helper functions called by the interpreter's builtin dispatch
|
||||
//!
|
||||
//! Developers never need to call anything here directly. The interpreter
|
||||
//! instruments everything automatically. Optional `log_*`, `trace_*`, and
|
||||
//! `metric_*` builtins are also wired through this module for explicit use.
|
||||
//!
|
||||
//! ## Graceful degradation
|
||||
//!
|
||||
//! If the OTLP endpoint is unreachable, a single warning is emitted to stderr
|
||||
//! on the first failure, then telemetry is silently dropped. Programs never
|
||||
//! fail because observability is down.
|
||||
|
||||
use std::sync::{
|
||||
OnceLock,
|
||||
atomic::{AtomicBool, Ordering},
|
||||
};
|
||||
use std::sync::mpsc::{self, SyncSender};
|
||||
use std::time::{SystemTime, UNIX_EPOCH};
|
||||
use std::collections::HashMap;
|
||||
|
||||
// ── Public re-export ──────────────────────────────────────────────────────────
|
||||
|
||||
pub use context::SpanGuard;
|
||||
|
||||
// ── Constants ─────────────────────────────────────────────────────────────────
|
||||
|
||||
const DEFAULT_OTLP_ENDPOINT: &str = "http://alloy.neuralplatform.ai:4318";
|
||||
const BATCH_SIZE: usize = 64;
|
||||
const BATCH_TIMEOUT_MS: u64 = 5_000;
|
||||
|
||||
// ── Telemetry payload types ───────────────────────────────────────────────────
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct Span {
|
||||
pub trace_id: String,
|
||||
pub span_id: String,
|
||||
pub parent_id: Option<String>,
|
||||
pub name: String,
|
||||
pub start_ns: u64,
|
||||
pub end_ns: u64,
|
||||
pub status: SpanStatus,
|
||||
pub attrs: Vec<(String, AttrValue)>,
|
||||
pub events: Vec<SpanEvent>,
|
||||
pub service: String,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
pub enum SpanStatus {
|
||||
Ok,
|
||||
Error(String),
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct SpanEvent {
|
||||
pub name: String,
|
||||
pub time_ns: u64,
|
||||
pub attrs: Vec<(String, AttrValue)>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub enum AttrValue {
|
||||
Str(String),
|
||||
Int(i64),
|
||||
Float(f64),
|
||||
Bool(bool),
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct LogRecord {
|
||||
pub time_ns: u64,
|
||||
pub severity: LogSeverity,
|
||||
pub body: String,
|
||||
pub attrs: Vec<(String, AttrValue)>,
|
||||
pub service: String,
|
||||
pub trace_id: Option<String>,
|
||||
pub span_id: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Copy)]
|
||||
pub enum LogSeverity {
|
||||
Debug,
|
||||
Info,
|
||||
Warn,
|
||||
Error,
|
||||
}
|
||||
|
||||
impl LogSeverity {
|
||||
fn number(self) -> u32 {
|
||||
match self {
|
||||
LogSeverity::Debug => 5,
|
||||
LogSeverity::Info => 9,
|
||||
LogSeverity::Warn => 13,
|
||||
LogSeverity::Error => 17,
|
||||
}
|
||||
}
|
||||
fn text(self) -> &'static str {
|
||||
match self {
|
||||
LogSeverity::Debug => "DEBUG",
|
||||
LogSeverity::Info => "INFO",
|
||||
LogSeverity::Warn => "WARN",
|
||||
LogSeverity::Error => "ERROR",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct Metric {
|
||||
pub name: String,
|
||||
pub kind: MetricKind,
|
||||
pub value: f64,
|
||||
pub attrs: Vec<(String, AttrValue)>,
|
||||
pub time_ns: u64,
|
||||
pub service: String,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub enum MetricKind {
|
||||
Counter,
|
||||
Gauge,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
enum TelemetryItem {
|
||||
Span(Span),
|
||||
Log(LogRecord),
|
||||
Metric(Metric),
|
||||
}
|
||||
|
||||
// ── Global telemetry sender ───────────────────────────────────────────────────
|
||||
|
||||
static TELEMETRY_TX: OnceLock<Option<SyncSender<TelemetryItem>>> = OnceLock::new();
|
||||
static OTLP_WARNED: AtomicBool = AtomicBool::new(false);
|
||||
|
||||
/// Initialise the telemetry background thread. Called once on interpreter start.
|
||||
/// `service_name` is the El program filename (without extension).
|
||||
pub fn init(service_name: &str) {
|
||||
let endpoint = std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT")
|
||||
.unwrap_or_else(|_| DEFAULT_OTLP_ENDPOINT.to_string());
|
||||
let svc = service_name.to_string();
|
||||
|
||||
// Bounded channel — if the exporter falls behind, new items are dropped.
|
||||
let (tx, rx) = mpsc::sync_channel::<TelemetryItem>(4096);
|
||||
|
||||
// Store the sender before starting the thread so callers can send immediately.
|
||||
TELEMETRY_TX.get_or_init(|| Some(tx));
|
||||
|
||||
std::thread::Builder::new()
|
||||
.name("el-telemetry".to_string())
|
||||
.spawn(move || {
|
||||
exporter_loop(rx, &endpoint, &svc);
|
||||
})
|
||||
.ok(); // If the thread fails to spawn, we degrade silently.
|
||||
}
|
||||
|
||||
/// Send one telemetry item. Never panics. Drops if channel is full or uninitialised.
|
||||
fn send(item: TelemetryItem) {
|
||||
if let Some(Some(tx)) = TELEMETRY_TX.get() {
|
||||
let _ = tx.try_send(item);
|
||||
}
|
||||
}
|
||||
|
||||
// ── Span builder / emitter ────────────────────────────────────────────────────
|
||||
|
||||
pub fn emit_span(span: Span) {
|
||||
send(TelemetryItem::Span(span));
|
||||
}
|
||||
|
||||
pub fn emit_log(record: LogRecord) {
|
||||
send(TelemetryItem::Log(record));
|
||||
}
|
||||
|
||||
pub fn emit_metric(metric: Metric) {
|
||||
send(TelemetryItem::Metric(metric));
|
||||
}
|
||||
|
||||
// ── Thread-local context ──────────────────────────────────────────────────────
|
||||
|
||||
pub mod context {
|
||||
use super::*;
|
||||
|
||||
thread_local! {
|
||||
/// Stack of active span IDs for the current thread.
|
||||
static SPAN_STACK: std::cell::RefCell<Vec<(String, String)>> =
|
||||
std::cell::RefCell::new(Vec::new());
|
||||
}
|
||||
|
||||
/// Get the current (innermost) span context: (trace_id, span_id).
|
||||
pub fn current_span() -> Option<(String, String)> {
|
||||
SPAN_STACK.with(|s| s.borrow().last().cloned())
|
||||
}
|
||||
|
||||
/// Push a span context onto the thread-local stack.
|
||||
pub fn push_span(trace_id: String, span_id: String) {
|
||||
SPAN_STACK.with(|s| s.borrow_mut().push((trace_id, span_id)));
|
||||
}
|
||||
|
||||
/// Pop the innermost span context.
|
||||
pub fn pop_span() {
|
||||
SPAN_STACK.with(|s| { s.borrow_mut().pop(); });
|
||||
}
|
||||
|
||||
/// A RAII guard that closes the span when it goes out of scope.
|
||||
pub struct SpanGuard {
|
||||
pub span: Span,
|
||||
}
|
||||
|
||||
impl SpanGuard {
|
||||
pub fn new(name: &str, service: &str) -> Self {
|
||||
let (trace_id, parent_id) = current_span()
|
||||
.map(|(t, s)| (t, Some(s)))
|
||||
.unwrap_or_else(|| (new_trace_id(), None));
|
||||
let span_id = new_span_id();
|
||||
push_span(trace_id.clone(), span_id.clone());
|
||||
SpanGuard {
|
||||
span: Span {
|
||||
trace_id,
|
||||
span_id,
|
||||
parent_id,
|
||||
name: name.to_string(),
|
||||
start_ns: now_ns(),
|
||||
end_ns: 0,
|
||||
status: SpanStatus::Ok,
|
||||
attrs: Vec::new(),
|
||||
events: Vec::new(),
|
||||
service: service.to_string(),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
pub fn attr(mut self, k: &str, v: AttrValue) -> Self {
|
||||
self.span.attrs.push((k.to_string(), v));
|
||||
self
|
||||
}
|
||||
|
||||
pub fn error(mut self, msg: &str) -> Self {
|
||||
self.span.status = SpanStatus::Error(msg.to_string());
|
||||
self.span.events.push(SpanEvent {
|
||||
name: "exception".to_string(),
|
||||
time_ns: now_ns(),
|
||||
attrs: vec![("exception.message".to_string(), AttrValue::Str(msg.to_string()))],
|
||||
});
|
||||
self
|
||||
}
|
||||
|
||||
/// Finish the span without dropping the guard (for manual control).
|
||||
pub fn finish(mut self) -> Span {
|
||||
pop_span();
|
||||
self.span.end_ns = now_ns();
|
||||
self.span.clone()
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for SpanGuard {
|
||||
fn drop(&mut self) {
|
||||
// Only pop if not already finished manually.
|
||||
// We detect this by checking if end_ns is still 0.
|
||||
if self.span.end_ns == 0 {
|
||||
pop_span();
|
||||
self.span.end_ns = now_ns();
|
||||
emit_span(self.span.clone());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ── ID generation ─────────────────────────────────────────────────────────────
|
||||
|
||||
pub fn new_trace_id() -> String {
|
||||
let id = uuid::Uuid::new_v4();
|
||||
hex::encode(id.as_bytes())
|
||||
}
|
||||
|
||||
pub fn new_span_id() -> String {
|
||||
let id = uuid::Uuid::new_v4();
|
||||
// Span ID is 8 bytes
|
||||
hex::encode(&id.as_bytes()[..8])
|
||||
}
|
||||
|
||||
// ── Timing ────────────────────────────────────────────────────────────────────
|
||||
|
||||
pub fn now_ns() -> u64 {
|
||||
SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.map(|d| d.as_nanos() as u64)
|
||||
.unwrap_or(0)
|
||||
}
|
||||
|
||||
pub fn now_ms() -> u64 {
|
||||
now_ns() / 1_000_000
|
||||
}
|
||||
|
||||
// ── Service name global ───────────────────────────────────────────────────────
|
||||
|
||||
static SERVICE_NAME: OnceLock<String> = OnceLock::new();
|
||||
|
||||
pub fn service_name() -> &'static str {
|
||||
SERVICE_NAME.get().map(|s| s.as_str()).unwrap_or("el-program")
|
||||
}
|
||||
|
||||
pub fn set_service_name(name: &str) {
|
||||
let _ = SERVICE_NAME.set(name.to_string());
|
||||
}
|
||||
|
||||
// ── High-level tracing helpers ────────────────────────────────────────────────
|
||||
|
||||
/// Instrument a function call. Returns a SpanGuard; emit it when done.
|
||||
pub fn start_fn_span(fn_name: &str) -> context::SpanGuard {
|
||||
context::SpanGuard::new(fn_name, service_name())
|
||||
}
|
||||
|
||||
/// Emit a log at the given severity.
|
||||
pub fn log(severity: LogSeverity, body: &str) {
|
||||
let (trace_id, span_id) = context::current_span()
|
||||
.map(|(t, s)| (Some(t), Some(s)))
|
||||
.unwrap_or((None, None));
|
||||
emit_log(LogRecord {
|
||||
time_ns: now_ns(),
|
||||
severity,
|
||||
body: body.to_string(),
|
||||
attrs: Vec::new(),
|
||||
service: service_name().to_string(),
|
||||
trace_id,
|
||||
span_id,
|
||||
});
|
||||
}
|
||||
|
||||
pub fn log_debug(msg: &str) { log(LogSeverity::Debug, msg); }
|
||||
pub fn log_info(msg: &str) { log(LogSeverity::Info, msg); }
|
||||
pub fn log_warn(msg: &str) { log(LogSeverity::Warn, msg); }
|
||||
pub fn log_error(msg: &str) { log(LogSeverity::Error, msg); }
|
||||
|
||||
/// Convenience: emit a metric counter.
|
||||
pub fn counter(name: &str, value: f64, attrs: Vec<(String, AttrValue)>) {
|
||||
emit_metric(Metric {
|
||||
name: name.to_string(),
|
||||
kind: MetricKind::Counter,
|
||||
value,
|
||||
attrs,
|
||||
time_ns: now_ns(),
|
||||
service: service_name().to_string(),
|
||||
});
|
||||
}
|
||||
|
||||
/// Convenience: emit a metric gauge.
|
||||
pub fn gauge(name: &str, value: f64, attrs: Vec<(String, AttrValue)>) {
|
||||
emit_metric(Metric {
|
||||
name: name.to_string(),
|
||||
kind: MetricKind::Gauge,
|
||||
value,
|
||||
attrs,
|
||||
time_ns: now_ns(),
|
||||
service: service_name().to_string(),
|
||||
});
|
||||
}
|
||||
|
||||
// ── OTLP/HTTP JSON exporter ───────────────────────────────────────────────────
|
||||
// Uses the OTLP/HTTP JSON format (not protobuf) which Alloy accepts.
|
||||
|
||||
fn exporter_loop(
|
||||
rx: mpsc::Receiver<TelemetryItem>,
|
||||
endpoint: &str,
|
||||
service: &str,
|
||||
) {
|
||||
let client = reqwest::blocking::Client::builder()
|
||||
.timeout(std::time::Duration::from_secs(10))
|
||||
.build()
|
||||
.unwrap_or_else(|_| reqwest::blocking::Client::new());
|
||||
|
||||
let spans_url = format!("{}/v1/traces", endpoint);
|
||||
let logs_url = format!("{}/v1/logs", endpoint);
|
||||
let metrics_url = format!("{}/v1/metrics", endpoint);
|
||||
|
||||
let mut spans: Vec<Span> = Vec::with_capacity(BATCH_SIZE);
|
||||
let mut logs: Vec<LogRecord> = Vec::with_capacity(BATCH_SIZE);
|
||||
let mut metrics: Vec<Metric> = Vec::with_capacity(BATCH_SIZE);
|
||||
|
||||
let timeout = std::time::Duration::from_millis(BATCH_TIMEOUT_MS);
|
||||
|
||||
loop {
|
||||
// Try to receive one item with a timeout, then drain available items.
|
||||
match rx.recv_timeout(timeout) {
|
||||
Ok(item) => {
|
||||
enqueue_item(item, &mut spans, &mut logs, &mut metrics);
|
||||
// Drain any immediately available items.
|
||||
while let Ok(item) = rx.try_recv() {
|
||||
enqueue_item(item, &mut spans, &mut logs, &mut metrics);
|
||||
if spans.len() + logs.len() + metrics.len() >= BATCH_SIZE {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(mpsc::RecvTimeoutError::Timeout) => {
|
||||
// Flush whatever we have.
|
||||
}
|
||||
Err(mpsc::RecvTimeoutError::Disconnected) => {
|
||||
// Channel closed — flush and exit.
|
||||
flush(&client, &spans_url, &logs_url, &metrics_url,
|
||||
service, &spans, &logs, &metrics);
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if !spans.is_empty() || !logs.is_empty() || !metrics.is_empty() {
|
||||
flush(&client, &spans_url, &logs_url, &metrics_url,
|
||||
service, &spans, &logs, &metrics);
|
||||
spans.clear();
|
||||
logs.clear();
|
||||
metrics.clear();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn enqueue_item(
|
||||
item: TelemetryItem,
|
||||
spans: &mut Vec<Span>,
|
||||
logs: &mut Vec<LogRecord>,
|
||||
metrics: &mut Vec<Metric>,
|
||||
) {
|
||||
match item {
|
||||
TelemetryItem::Span(s) => spans.push(s),
|
||||
TelemetryItem::Log(l) => logs.push(l),
|
||||
TelemetryItem::Metric(m) => metrics.push(m),
|
||||
}
|
||||
}
|
||||
|
||||
fn flush(
|
||||
client: &reqwest::blocking::Client,
|
||||
spans_url: &str,
|
||||
logs_url: &str,
|
||||
metrics_url: &str,
|
||||
service: &str,
|
||||
spans: &[Span],
|
||||
logs: &[LogRecord],
|
||||
metrics: &[Metric],
|
||||
) {
|
||||
if !spans.is_empty() {
|
||||
let body = build_traces_json(service, spans);
|
||||
post_otlp(client, spans_url, &body);
|
||||
}
|
||||
if !logs.is_empty() {
|
||||
let body = build_logs_json(service, logs);
|
||||
post_otlp(client, logs_url, &body);
|
||||
}
|
||||
if !metrics.is_empty() {
|
||||
let body = build_metrics_json(service, metrics);
|
||||
post_otlp(client, metrics_url, &body);
|
||||
}
|
||||
}
|
||||
|
||||
fn post_otlp(client: &reqwest::blocking::Client, url: &str, body: &str) {
|
||||
let res = client.post(url)
|
||||
.header("Content-Type", "application/json")
|
||||
.body(body.to_string())
|
||||
.send();
|
||||
match res {
|
||||
Ok(r) if r.status().is_success() => {}
|
||||
Ok(r) => {
|
||||
if !OTLP_WARNED.swap(true, Ordering::Relaxed) {
|
||||
eprintln!("[el-telemetry] OTLP export failed: HTTP {}", r.status());
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
if !OTLP_WARNED.swap(true, Ordering::Relaxed) {
|
||||
eprintln!("[el-telemetry] OTLP endpoint unreachable ({}), telemetry will be dropped", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ── JSON serialisation for OTLP/HTTP ─────────────────────────────────────────
|
||||
|
||||
fn attr_value_json(v: &AttrValue) -> serde_json::Value {
|
||||
match v {
|
||||
AttrValue::Str(s) => serde_json::json!({"stringValue": s}),
|
||||
AttrValue::Int(n) => serde_json::json!({"intValue": n.to_string()}),
|
||||
AttrValue::Float(f) => serde_json::json!({"doubleValue": f}),
|
||||
AttrValue::Bool(b) => serde_json::json!({"boolValue": b}),
|
||||
}
|
||||
}
|
||||
|
||||
fn attrs_json(attrs: &[(String, AttrValue)]) -> serde_json::Value {
|
||||
serde_json::Value::Array(
|
||||
attrs.iter().map(|(k, v)| {
|
||||
serde_json::json!({"key": k, "value": attr_value_json(v)})
|
||||
}).collect()
|
||||
)
|
||||
}
|
||||
|
||||
fn resource_json(service: &str) -> serde_json::Value {
|
||||
serde_json::json!({
|
||||
"attributes": [
|
||||
{"key": "service.name", "value": {"stringValue": service}}
|
||||
]
|
||||
})
|
||||
}
|
||||
|
||||
fn build_traces_json(service: &str, spans: &[Span]) -> String {
|
||||
// Group spans by trace_id for correct OTLP structure
|
||||
let mut by_trace: HashMap<&str, Vec<serde_json::Value>> = HashMap::new();
|
||||
for span in spans {
|
||||
let js = span_json(span);
|
||||
by_trace.entry(&span.trace_id).or_default().push(js);
|
||||
}
|
||||
|
||||
let scope_spans: Vec<serde_json::Value> = by_trace.values().map(|sps| {
|
||||
serde_json::json!({
|
||||
"scope": {"name": "el-runtime", "version": "0.1.0"},
|
||||
"spans": sps
|
||||
})
|
||||
}).collect();
|
||||
|
||||
let payload = serde_json::json!({
|
||||
"resourceSpans": [{
|
||||
"resource": resource_json(service),
|
||||
"scopeSpans": scope_spans
|
||||
}]
|
||||
});
|
||||
payload.to_string()
|
||||
}
|
||||
|
||||
fn span_json(span: &Span) -> serde_json::Value {
|
||||
let status = match &span.status {
|
||||
SpanStatus::Ok => serde_json::json!({"code": 1}),
|
||||
SpanStatus::Error(msg) => serde_json::json!({"code": 2, "message": msg}),
|
||||
};
|
||||
let events: Vec<serde_json::Value> = span.events.iter().map(|e| {
|
||||
serde_json::json!({
|
||||
"name": e.name,
|
||||
"timeUnixNano": e.time_ns.to_string(),
|
||||
"attributes": attrs_json(&e.attrs)
|
||||
})
|
||||
}).collect();
|
||||
|
||||
let mut js = serde_json::json!({
|
||||
"traceId": span.trace_id,
|
||||
"spanId": span.span_id,
|
||||
"name": span.name,
|
||||
"startTimeUnixNano": span.start_ns.to_string(),
|
||||
"endTimeUnixNano": span.end_ns.to_string(),
|
||||
"attributes": attrs_json(&span.attrs),
|
||||
"events": events,
|
||||
"status": status,
|
||||
"kind": 1 // INTERNAL
|
||||
});
|
||||
if let Some(pid) = &span.parent_id {
|
||||
js["parentSpanId"] = serde_json::Value::String(pid.clone());
|
||||
}
|
||||
js
|
||||
}
|
||||
|
||||
fn build_logs_json(service: &str, logs: &[LogRecord]) -> String {
|
||||
let records: Vec<serde_json::Value> = logs.iter().map(|l| {
|
||||
let mut r = serde_json::json!({
|
||||
"timeUnixNano": l.time_ns.to_string(),
|
||||
"severityNumber": l.severity.number(),
|
||||
"severityText": l.severity.text(),
|
||||
"body": {"stringValue": l.body},
|
||||
"attributes": attrs_json(&l.attrs)
|
||||
});
|
||||
if let Some(tid) = &l.trace_id {
|
||||
r["traceId"] = serde_json::Value::String(tid.clone());
|
||||
}
|
||||
if let Some(sid) = &l.span_id {
|
||||
r["spanId"] = serde_json::Value::String(sid.clone());
|
||||
}
|
||||
r
|
||||
}).collect();
|
||||
|
||||
let payload = serde_json::json!({
|
||||
"resourceLogs": [{
|
||||
"resource": resource_json(service),
|
||||
"scopeLogs": [{
|
||||
"scope": {"name": "el-runtime", "version": "0.1.0"},
|
||||
"logRecords": records
|
||||
}]
|
||||
}]
|
||||
});
|
||||
payload.to_string()
|
||||
}
|
||||
|
||||
fn build_metrics_json(service: &str, metrics: &[Metric]) -> String {
|
||||
let metric_items: Vec<serde_json::Value> = metrics.iter().map(|m| {
|
||||
let data_point = serde_json::json!({
|
||||
"timeUnixNano": m.time_ns.to_string(),
|
||||
"asDouble": m.value,
|
||||
"attributes": attrs_json(&m.attrs)
|
||||
});
|
||||
match m.kind {
|
||||
MetricKind::Counter => serde_json::json!({
|
||||
"name": m.name,
|
||||
"sum": {
|
||||
"dataPoints": [data_point],
|
||||
"aggregationTemporality": 2, // CUMULATIVE
|
||||
"isMonotonic": true
|
||||
}
|
||||
}),
|
||||
MetricKind::Gauge => serde_json::json!({
|
||||
"name": m.name,
|
||||
"gauge": {
|
||||
"dataPoints": [data_point]
|
||||
}
|
||||
}),
|
||||
}
|
||||
}).collect();
|
||||
|
||||
let payload = serde_json::json!({
|
||||
"resourceMetrics": [{
|
||||
"resource": resource_json(service),
|
||||
"scopeMetrics": [{
|
||||
"scope": {"name": "el-runtime", "version": "0.1.0"},
|
||||
"metrics": metric_items
|
||||
}]
|
||||
}]
|
||||
});
|
||||
payload.to_string()
|
||||
}
|
||||
|
||||
// ── parse_tags_string: "k=v,k2=v2" → Vec<(String, AttrValue)> ────────────────
|
||||
|
||||
pub fn parse_tags_string(tags: &str) -> Vec<(String, AttrValue)> {
|
||||
if tags.is_empty() {
|
||||
return Vec::new();
|
||||
}
|
||||
tags.split(',')
|
||||
.filter_map(|pair| {
|
||||
let mut parts = pair.splitn(2, '=');
|
||||
let k = parts.next()?.trim().to_string();
|
||||
let v = parts.next().unwrap_or("").trim().to_string();
|
||||
if k.is_empty() {
|
||||
None
|
||||
} else {
|
||||
Some((k, AttrValue::Str(v)))
|
||||
}
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
@@ -0,0 +1,20 @@
|
||||
[package]
|
||||
name = "elvm"
|
||||
description = "El Virtual Machine — executes compiled El bytecode (.elc) natively"
|
||||
version.workspace = true
|
||||
edition.workspace = true
|
||||
license.workspace = true
|
||||
|
||||
[[bin]]
|
||||
name = "elvm"
|
||||
path = "src/main.rs"
|
||||
|
||||
[dependencies]
|
||||
el-compiler = { workspace = true }
|
||||
el-vm = { workspace = true }
|
||||
clap = { workspace = true }
|
||||
|
||||
# Native window / WebView (macOS: WKWebView via wry)
|
||||
wry = { version = "0.47", default-features = false }
|
||||
winit = { version = "0.29", default-features = false, features = ["rwh_05", "rwh_06"] }
|
||||
dpi = "0.1"
|
||||
@@ -0,0 +1,146 @@
|
||||
//! elvm — El Virtual Machine
|
||||
//!
|
||||
//! The standalone El VM binary. Loads and executes compiled El bytecode (.elc)
|
||||
//! files produced by `el compile` or `el build-file`.
|
||||
//!
|
||||
//! # Usage
|
||||
//!
|
||||
//! elvm <file.elc> [args...]
|
||||
//! elvm --version
|
||||
//! elvm --help
|
||||
//!
|
||||
//! If the environment variable `NEURON_WINDOW_URL` is set, elvm opens a native
|
||||
//! macOS window (WKWebView via wry) at that URL instead of executing bytecode.
|
||||
//! This allows UI apps to be launched as proper desktop windows:
|
||||
//!
|
||||
//! NEURON_WINDOW_URL="http://localhost:7749" elvm dist/neuron-ui.elc
|
||||
|
||||
use clap::Parser;
|
||||
use std::path::PathBuf;
|
||||
|
||||
#[derive(Parser, Debug)]
|
||||
#[command(
|
||||
name = "elvm",
|
||||
about = "El Virtual Machine — execute compiled El bytecode (.elc)",
|
||||
long_about = "The El VM is the native El execution substrate.\n\
|
||||
Run .elc files produced by `el compile` or `el build-file`.\n\n\
|
||||
Set NEURON_WINDOW_URL=<url> to open a native WebView window instead.",
|
||||
version
|
||||
)]
|
||||
struct Cli {
|
||||
/// Compiled El bytecode file to execute (*.elc).
|
||||
file: PathBuf,
|
||||
|
||||
/// Arguments forwarded to the program (accessible via `args()`).
|
||||
#[arg(trailing_var_arg = true)]
|
||||
args: Vec<String>,
|
||||
}
|
||||
|
||||
fn main() {
|
||||
let cli = Cli::parse();
|
||||
|
||||
// If NEURON_WINDOW_URL is set, open a native WebView window at that URL.
|
||||
if let Ok(url) = std::env::var("NEURON_WINDOW_URL") {
|
||||
if let Err(e) = open_window(&url) {
|
||||
eprintln!("elvm: window error: {e}");
|
||||
std::process::exit(1);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
if let Err(e) = run(cli) {
|
||||
eprintln!("elvm: error: {e}");
|
||||
std::process::exit(1);
|
||||
}
|
||||
}
|
||||
|
||||
fn run(cli: Cli) -> Result<(), Box<dyn std::error::Error>> {
|
||||
let bytes = std::fs::read(&cli.file)
|
||||
.map_err(|e| format!("cannot read {}: {e}", cli.file.display()))?;
|
||||
|
||||
let instructions = el_compiler::Bytecode::deserialize_all(&bytes)
|
||||
.map_err(|e| format!("cannot load bytecode from {}: {e}", cli.file.display()))?;
|
||||
|
||||
// Detect format and print diagnostic.
|
||||
let is_elvm_container = bytes.starts_with(el_compiler::ELVM_MAGIC);
|
||||
if is_elvm_container {
|
||||
// Normal path — ELVM container.
|
||||
} else {
|
||||
eprintln!("elvm: warning: {} does not have an ELVM header — treating as legacy JSON bytecode", cli.file.display());
|
||||
}
|
||||
|
||||
let mut vm = el_vm::ElVm::new();
|
||||
vm.run(&instructions, &cli.args);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Opens a native WebView window at `url`.
|
||||
///
|
||||
/// On macOS: uses wry (WKWebView) + winit for a proper native desktop window.
|
||||
/// On other platforms: prints the URL (fallback).
|
||||
fn open_window(url: &str) -> Result<(), Box<dyn std::error::Error>> {
|
||||
#[cfg(target_os = "macos")]
|
||||
return open_native_window(url);
|
||||
|
||||
#[cfg(not(target_os = "macos"))]
|
||||
{
|
||||
eprintln!("elvm: native window not supported on this platform");
|
||||
println!("elvm: open {url} in your browser");
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(target_os = "macos")]
|
||||
fn open_native_window(url: &str) -> Result<(), Box<dyn std::error::Error>> {
|
||||
use dpi::LogicalSize;
|
||||
use winit::{
|
||||
event::{Event, WindowEvent},
|
||||
event_loop::{ControlFlow, EventLoop},
|
||||
window::WindowBuilder,
|
||||
};
|
||||
use wry::{Rect, WebViewBuilder};
|
||||
|
||||
let event_loop = EventLoop::new().map_err(|e| format!("event loop: {e}"))?;
|
||||
|
||||
let window = WindowBuilder::new()
|
||||
.with_title("Neuron")
|
||||
.with_inner_size(winit::dpi::LogicalSize::new(1600u32, 1000u32))
|
||||
.with_min_inner_size(winit::dpi::LogicalSize::new(900u32, 600u32))
|
||||
.with_resizable(true)
|
||||
.build(&event_loop)
|
||||
.map_err(|e| format!("window: {e}"))?;
|
||||
|
||||
let url_owned = url.to_string();
|
||||
let webview = WebViewBuilder::new()
|
||||
.with_url(&url_owned)
|
||||
.build_as_child(&window)
|
||||
.map_err(|e| format!("webview: {e}"))?;
|
||||
|
||||
event_loop
|
||||
.run(move |event, evl| {
|
||||
evl.set_control_flow(ControlFlow::Wait);
|
||||
|
||||
match event {
|
||||
Event::WindowEvent {
|
||||
event: WindowEvent::Resized(size),
|
||||
..
|
||||
} => {
|
||||
let scale = window.scale_factor();
|
||||
let logical = size.to_logical::<u32>(scale);
|
||||
let _ = webview.set_bounds(Rect {
|
||||
position: dpi::LogicalPosition::new(0, 0).into(),
|
||||
size: LogicalSize::new(logical.width, logical.height).into(),
|
||||
});
|
||||
}
|
||||
Event::WindowEvent {
|
||||
event: WindowEvent::CloseRequested,
|
||||
..
|
||||
} => evl.exit(),
|
||||
_ => {}
|
||||
}
|
||||
})
|
||||
.map_err(|e| format!("event loop run: {e}"))?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
Reference in New Issue
Block a user