use std::collections::HashMap;
use std::future::Future;
use std::pin::Pin;
use std::sync::OnceLock;
use std::time::Instant;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use crate::ast::{HttpMethod, ResolverType};
use crate::compiler::MultiEntityBytecode;
use crate::resolvers::{TokenMetadataResolverClient, UrlResolverClient};
use crate::vm::{
CachedResolverValue, ResolverRequest, TokenResolverMissAction, UpdateContext, VmContext,
};
use crate::Mutation;
pub type ResolverBatchResult =
Result<HashMap<String, Value>, Box<dyn std::error::Error + Send + Sync>>;
pub type ResolverBatchFuture<'a> = Pin<Box<dyn Future<Output = ResolverBatchResult> + Send + 'a>>;
pub type ResolverApplyFuture<'a> = Pin<Box<dyn Future<Output = Vec<Mutation>> + Send + 'a>>;
pub type SharedRuntimeResolver = std::sync::Arc<dyn RuntimeResolver>;
#[derive(Clone, Copy)]
struct RuntimeResolverProfileConfig {
enabled: bool,
slow_ms: f64,
}
fn runtime_resolver_profile_config() -> RuntimeResolverProfileConfig {
static CONFIG: OnceLock<RuntimeResolverProfileConfig> = OnceLock::new();
*CONFIG.get_or_init(|| RuntimeResolverProfileConfig {
enabled: std::env::var("ARETE_PROFILE_CODEPATHS")
.ok()
.or_else(|| std::env::var("ARETE_PROFILE_RUNTIME_RESOLVER").ok())
.map(|value| {
matches!(
value.to_ascii_lowercase().as_str(),
"1" | "true" | "yes" | "on"
)
})
.unwrap_or(false),
slow_ms: std::env::var("ARETE_PROFILE_CODEPATHS_SLOW_MS")
.ok()
.or_else(|| std::env::var("ARETE_PROFILE_RUNTIME_RESOLVER_SLOW_MS").ok())
.and_then(|value| value.parse::<f64>().ok())
.unwrap_or(5.0),
})
}
fn elapsed_ms(start: Instant) -> f64 {
start.elapsed().as_secs_f64() * 1000.0
}
fn should_log_runtime_resolver_profile(total_ms: f64) -> bool {
let config = runtime_resolver_profile_config();
config.enabled && total_ms >= config.slow_ms
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Hash)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum RuntimeResolverRequest {
TokenMetadata {
key: String,
mint: String,
},
UrlJson {
key: String,
url: String,
method: HttpMethod,
},
}
impl RuntimeResolverRequest {
pub fn key(&self) -> &str {
match self {
Self::TokenMetadata { key, .. } | Self::UrlJson { key, .. } => key,
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RuntimeResolverBatchRequest {
pub requests: Vec<RuntimeResolverRequest>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RuntimeResolverResponse {
pub key: String,
pub value: Value,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RuntimeResolverBatchResponse {
#[serde(default)]
pub resolved: Vec<RuntimeResolverResponse>,
}
#[derive(Debug)]
struct PendingRuntimeResolverRequest {
request: ResolverRequest,
backend_request: RuntimeResolverRequest,
}
pub trait RuntimeResolver: Send + Sync {
fn resolve_batch<'a>(
&'a self,
requests: &'a [RuntimeResolverRequest],
) -> ResolverBatchFuture<'a>;
fn resolve_and_apply<'a>(
&'a self,
vm: &'a std::sync::Mutex<VmContext>,
bytecode: &'a MultiEntityBytecode,
requests: Vec<ResolverRequest>,
apply_context: Option<UpdateContext>,
) -> ResolverApplyFuture<'a> {
Box::pin(async move {
let total_started_at = Instant::now();
if requests.is_empty() {
return Vec::new();
}
let request_count = requests.len();
let mut cached = Vec::new();
let mut pending = Vec::new();
let mut invalid = Vec::new();
let mut negative_cached = Vec::new();
let classify_started_at = Instant::now();
let (
classify_ms,
cached_count,
pending_count,
invalid_count,
negative_cached_count,
unique_request_count,
) = {
let mut vm_guard = vm.lock().unwrap_or_else(|e| e.into_inner());
for request in requests {
let canonical_key =
runtime_resolver_cache_key(&request.resolver, &request.input);
if let Some(cached_value) = vm_guard.get_cached_resolver_value(&canonical_key) {
match cached_value {
CachedResolverValue::Resolved(resolved_value) => {
cached.push((request, resolved_value));
}
CachedResolverValue::Negative => {
negative_cached.push(request);
}
}
continue;
}
match runtime_request_from_vm_request(&request) {
Some(backend_request) => pending.push(PendingRuntimeResolverRequest {
request,
backend_request,
}),
None => invalid.push(request),
}
}
let invalid_count = invalid.len();
if !invalid.is_empty() {
vm_guard.restore_resolver_requests(invalid);
}
let classify_ms = elapsed_ms(classify_started_at);
let cached_count = cached.len();
let pending_count = pending.len();
let negative_cached_count = negative_cached.len();
let unique_request_count = if pending.is_empty() {
0
} else {
let mut unique = HashMap::new();
for entry in &pending {
unique
.entry(entry.backend_request.key().to_string())
.or_insert_with(|| entry.backend_request.clone());
}
unique.len()
};
(
classify_ms,
cached_count,
pending_count,
invalid_count,
negative_cached_count,
unique_request_count,
)
};
let resolve_batch_started_at = Instant::now();
let resolved_map = if pending.is_empty() {
Ok(HashMap::new())
} else {
let mut unique = HashMap::new();
for entry in &pending {
unique
.entry(entry.backend_request.key().to_string())
.or_insert_with(|| entry.backend_request.clone());
}
let unique_requests: Vec<RuntimeResolverRequest> = unique.into_values().collect();
self.resolve_batch(&unique_requests).await
};
let resolve_batch_ms = elapsed_ms(resolve_batch_started_at);
let mut mutations = Vec::new();
let mut failed = Vec::new();
let mut retry_scheduled_count = 0usize;
let mut negative_cached_miss_count = 0usize;
let apply_cached_started_at = Instant::now();
let mut vm_guard = vm.lock().unwrap_or_else(|e| e.into_inner());
let previous_context = if apply_context.is_some() {
let previous = vm_guard.current_context().cloned();
vm_guard.set_current_context(apply_context.clone());
previous
} else {
None
};
for (request, resolved_value) in cached {
match vm_guard.apply_resolver_result(bytecode, &request.cache_key, resolved_value) {
Ok(mut new_mutations) => mutations.append(&mut new_mutations),
Err(err) => {
tracing::warn!(
cache_key = %request.cache_key,
error = %err,
"Failed to apply cached resolver result"
);
failed.push(request);
}
}
}
let apply_cached_ms = elapsed_ms(apply_cached_started_at);
let apply_pending_started_at = Instant::now();
match resolved_map {
Ok(resolved_map) => {
for entry in pending {
match resolved_map.get(entry.backend_request.key()) {
Some(resolved_value) => match vm_guard.apply_resolver_result(
bytecode,
&entry.request.cache_key,
resolved_value.clone(),
) {
Ok(mut new_mutations) => mutations.append(&mut new_mutations),
Err(err) => {
tracing::warn!(
cache_key = %entry.request.cache_key,
error = %err,
"Failed to apply resolver result"
);
failed.push(entry.request);
}
},
None => match entry.backend_request {
RuntimeResolverRequest::TokenMetadata { .. } => {
match vm_guard
.handle_token_resolver_miss(&entry.request.cache_key)
{
TokenResolverMissAction::RetryScheduled => {
retry_scheduled_count += 1;
}
TokenResolverMissAction::NegativeCached => {
negative_cached_miss_count += 1;
}
TokenResolverMissAction::Dropped => {}
}
}
RuntimeResolverRequest::UrlJson { .. } => {
failed.push(entry.request)
}
},
}
}
}
Err(err) => {
tracing::warn!(error = %err, "Runtime resolver backend request failed");
for entry in pending {
match entry.backend_request {
RuntimeResolverRequest::TokenMetadata { .. } => {
match vm_guard.handle_token_resolver_miss(&entry.request.cache_key)
{
TokenResolverMissAction::RetryScheduled => {
retry_scheduled_count += 1;
}
TokenResolverMissAction::NegativeCached => {
negative_cached_miss_count += 1;
}
TokenResolverMissAction::Dropped => {}
}
}
RuntimeResolverRequest::UrlJson { .. } => failed.push(entry.request),
}
}
}
}
let apply_pending_ms = elapsed_ms(apply_pending_started_at);
let failed_count = failed.len();
let restore_failed_started_at = Instant::now();
if !failed.is_empty() {
vm_guard.restore_resolver_requests(failed);
}
let restore_failed_ms = elapsed_ms(restore_failed_started_at);
if apply_context.is_some() {
vm_guard.set_current_context(previous_context);
}
let total_ms = elapsed_ms(total_started_at);
if should_log_runtime_resolver_profile(total_ms) {
tracing::info!(
stage = "runtime.resolver_apply",
request_count,
cached_count,
pending_count,
invalid_count,
negative_cached_count,
unique_request_count,
failed_count,
retry_scheduled_count,
negative_cached_miss_count,
mutation_count = mutations.len(),
total_ms,
classify_ms,
resolve_batch_ms,
apply_cached_ms,
apply_pending_ms,
restore_failed_ms,
"Runtime resolver profile"
);
}
mutations
})
}
}
pub struct InProcessResolver {
token_client: Option<TokenMetadataResolverClient>,
url_client: UrlResolverClient,
}
impl InProcessResolver {
pub fn from_env() -> Result<Self, Box<dyn std::error::Error + Send + Sync>> {
Ok(Self {
token_client: TokenMetadataResolverClient::from_env()?,
url_client: UrlResolverClient::new(),
})
}
pub fn new(
token_client: Option<TokenMetadataResolverClient>,
url_client: UrlResolverClient,
) -> Self {
Self {
token_client,
url_client,
}
}
pub async fn resolve_batch_internal(
&self,
requests: &[RuntimeResolverRequest],
) -> ResolverBatchResult {
let total_started_at = Instant::now();
let mut results = HashMap::new();
let mut token_requests = Vec::new();
let mut url_requests = Vec::new();
let classify_started_at = Instant::now();
for request in requests {
match request {
RuntimeResolverRequest::TokenMetadata { key, mint } => {
token_requests.push((key.clone(), mint.clone()));
}
RuntimeResolverRequest::UrlJson { key, url, method } => {
url_requests.push((key.clone(), url.clone(), method.clone()));
}
}
}
let classify_ms = elapsed_ms(classify_started_at);
let token_request_count = token_requests.len();
let url_request_count = url_requests.len();
let mut token_batch_ms = 0.0_f64;
let mut url_dedupe_ms = 0.0_f64;
let mut url_batch_ms = 0.0_f64;
let mut url_unique_request_count = 0usize;
if !token_requests.is_empty() {
let token_started_at = Instant::now();
if let Some(token_client) = &self.token_client {
let mints: Vec<String> = token_requests
.iter()
.map(|(_, mint)| mint.clone())
.collect();
match token_client.resolve_token_metadata(&mints).await {
Ok(resolved) => {
for (key, mint) in token_requests {
if let Some(value) = resolved.get(&mint) {
results.insert(key, value.clone());
}
}
}
Err(err) => {
tracing::warn!(error = %err, "Failed to resolve token metadata batch");
}
}
} else {
tracing::warn!(
count = token_requests.len(),
"DAS_API_ENDPOINT not set; token resolver requests will be re-queued"
);
}
token_batch_ms = elapsed_ms(token_started_at);
}
if !url_requests.is_empty() {
let url_dedupe_started_at = Instant::now();
let mut unique = HashMap::new();
for (key, url, method) in &url_requests {
unique
.entry((url.clone(), method.clone()))
.or_insert_with(Vec::new)
.push(key.clone());
}
url_unique_request_count = unique.len();
let batch_input: Vec<(String, HttpMethod)> = unique.keys().cloned().collect();
url_dedupe_ms = elapsed_ms(url_dedupe_started_at);
let url_batch_started_at = Instant::now();
let resolved = self.url_client.resolve_batch(&batch_input).await;
url_batch_ms = elapsed_ms(url_batch_started_at);
for ((url, method), keys) in unique {
if let Some(value) = resolved.get(&(url, method)) {
for key in keys {
results.insert(key, value.clone());
}
}
}
}
let total_ms = elapsed_ms(total_started_at);
if should_log_runtime_resolver_profile(total_ms) {
tracing::info!(
stage = "runtime.resolver_batch",
request_count = requests.len(),
token_request_count,
url_request_count,
url_unique_request_count,
resolved_count = results.len(),
total_ms,
classify_ms,
token_batch_ms,
url_dedupe_ms,
url_batch_ms,
"Runtime resolver batch profile"
);
}
Ok(results)
}
}
impl RuntimeResolver for InProcessResolver {
fn resolve_batch<'a>(
&'a self,
requests: &'a [RuntimeResolverRequest],
) -> ResolverBatchFuture<'a> {
Box::pin(async move { self.resolve_batch_internal(requests).await })
}
}
pub fn runtime_resolver_cache_key(resolver: &ResolverType, input: &Value) -> String {
crate::vm::resolver_cache_key(resolver, input)
}
fn runtime_request_from_vm_request(request: &ResolverRequest) -> Option<RuntimeResolverRequest> {
match &request.resolver {
ResolverType::Token => extract_mint_from_input(&request.input).map(|mint| {
RuntimeResolverRequest::TokenMetadata {
key: request.cache_key.clone(),
mint,
}
}),
ResolverType::Url(config) => match &request.input {
Value::String(url) if !url.is_empty() => Some(RuntimeResolverRequest::UrlJson {
key: request.cache_key.clone(),
url: url.clone(),
method: config.method.clone(),
}),
_ => None,
},
}
}
fn extract_mint_from_input(input: &Value) -> Option<String> {
match input {
Value::String(value) if !value.is_empty() => Some(value.clone()),
Value::Object(map) => map
.get("mint")
.and_then(|value| value.as_str())
.filter(|value| !value.is_empty())
.map(str::to_string),
_ => None,
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn token_request_extracts_mint_from_object_input() {
let request = ResolverRequest {
cache_key: "token:mint".to_string(),
resolver: ResolverType::Token,
input: serde_json::json!({ "mint": "abc" }),
};
let runtime_request = runtime_request_from_vm_request(&request).unwrap();
assert_eq!(
runtime_request,
RuntimeResolverRequest::TokenMetadata {
key: "token:mint".to_string(),
mint: "abc".to_string(),
}
);
}
#[test]
fn url_request_uses_existing_cache_key() {
let request = ResolverRequest {
cache_key: "url:get:https://example.com".to_string(),
resolver: ResolverType::Url(crate::ast::UrlResolverConfig {
url_source: crate::ast::UrlSource::FieldPath("metadata_url".to_string()),
method: HttpMethod::Get,
extract_path: None,
}),
input: serde_json::json!("https://example.com"),
};
let runtime_request = runtime_request_from_vm_request(&request).unwrap();
assert_eq!(
runtime_request,
RuntimeResolverRequest::UrlJson {
key: "url:get:https://example.com".to_string(),
url: "https://example.com".to_string(),
method: HttpMethod::Get,
}
);
}
}