Split runtime response cache policy
This commit is contained in:
@@ -1,21 +1,23 @@
|
||||
use std::sync::Arc;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
|
||||
use crank_core::{
|
||||
AdapterRegistry, CachedHeader, CachedResponse, CoordinationStateStore, ExecutionMode,
|
||||
HttpMethod, InvocationStatus, MeteringEvent, ResponseCacheStore, SharedMeteringSink,
|
||||
SharedProtocolAdapter, Target,
|
||||
AdapterRegistry, CoordinationStateStore, ExecutionMode, InvocationStatus, MeteringEvent,
|
||||
ResponseCacheStore, SharedMeteringSink, SharedProtocolAdapter,
|
||||
};
|
||||
use serde_json::{Map, Value, json};
|
||||
use sha2::{Digest, Sha256};
|
||||
use time::OffsetDateTime;
|
||||
use tokio::sync::{OwnedSemaphorePermit, Semaphore};
|
||||
use tracing::debug;
|
||||
|
||||
use crate::{
|
||||
AdapterResponse, PreparedRequest, ResolvedAuth, RuntimeError, RuntimeLimits, RuntimeOperation,
|
||||
RuntimeRequestContext, request_preparation::adapter_prepared_request,
|
||||
RuntimeRequestContext,
|
||||
request_preparation::adapter_prepared_request,
|
||||
response_cache::{
|
||||
adapter_response_from_cached, cached_response_from_adapter, response_cache_key,
|
||||
response_cache_ttl,
|
||||
},
|
||||
};
|
||||
|
||||
#[derive(Clone)]
|
||||
@@ -523,80 +525,3 @@ fn log_runtime_event(
|
||||
"runtime execution"
|
||||
);
|
||||
}
|
||||
|
||||
fn response_cache_ttl(operation: &RuntimeOperation) -> Option<Duration> {
|
||||
let is_cacheable_protocol =
|
||||
matches!(&operation.target, Target::Rest(target) if target.method == HttpMethod::Get);
|
||||
if !is_cacheable_protocol {
|
||||
return None;
|
||||
}
|
||||
|
||||
let policy = operation.execution_config.response_cache.as_ref()?;
|
||||
if operation.execution_config.auth_profile_ref.is_some() || policy.ttl_ms == 0 {
|
||||
return None;
|
||||
}
|
||||
|
||||
Some(Duration::from_millis(policy.ttl_ms))
|
||||
}
|
||||
|
||||
fn response_cache_key(
|
||||
operation: &RuntimeOperation,
|
||||
prepared_request: &PreparedRequest,
|
||||
request_context: Option<&RuntimeRequestContext>,
|
||||
) -> Option<String> {
|
||||
let _ = response_cache_ttl(operation)?;
|
||||
let scope = request_context?.response_cache_scope()?;
|
||||
let fingerprint = json!({
|
||||
"path_params": prepared_request.path_params,
|
||||
"query_params": prepared_request.query_params,
|
||||
"headers": prepared_request.headers,
|
||||
"body": prepared_request.body,
|
||||
});
|
||||
let fingerprint_bytes = serde_json::to_vec(&fingerprint).ok()?;
|
||||
let fingerprint_hash = URL_SAFE_NO_PAD.encode(Sha256::digest(fingerprint_bytes));
|
||||
|
||||
Some(format!(
|
||||
"crank:response:workspace:{}:agent:{}:operation:{}:version:{}:request:{}",
|
||||
scope.workspace_key,
|
||||
scope.agent_key,
|
||||
operation.operation_id.as_str(),
|
||||
operation.operation_version,
|
||||
fingerprint_hash
|
||||
))
|
||||
}
|
||||
|
||||
fn cached_response_from_adapter(adapter_response: &AdapterResponse) -> Option<CachedResponse> {
|
||||
let body = serde_json::to_vec(&adapter_response.body).ok()?;
|
||||
Some(CachedResponse {
|
||||
status: adapter_response.status_code,
|
||||
headers: adapter_response
|
||||
.headers
|
||||
.iter()
|
||||
.map(|(name, value)| CachedHeader {
|
||||
name: name.clone(),
|
||||
value: value.clone(),
|
||||
})
|
||||
.collect(),
|
||||
body,
|
||||
data: serde_json::to_vec(&adapter_response.data).ok()?,
|
||||
})
|
||||
}
|
||||
|
||||
fn adapter_response_from_cached(
|
||||
cached_response: CachedResponse,
|
||||
) -> Result<AdapterResponse, String> {
|
||||
let body: Value =
|
||||
serde_json::from_slice(&cached_response.body).map_err(|error| error.to_string())?;
|
||||
let data: Value =
|
||||
serde_json::from_slice(&cached_response.data).map_err(|error| error.to_string())?;
|
||||
Ok(AdapterResponse {
|
||||
status_code: cached_response.status,
|
||||
headers: cached_response
|
||||
.headers
|
||||
.into_iter()
|
||||
.map(|header| (header.name, header.value))
|
||||
.collect(),
|
||||
body,
|
||||
data,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -11,6 +11,7 @@ mod model;
|
||||
mod rate_limit;
|
||||
mod request_context;
|
||||
mod request_preparation;
|
||||
mod response_cache;
|
||||
mod secret_crypto;
|
||||
|
||||
pub use auth::ResolvedAuth;
|
||||
|
||||
@@ -0,0 +1,87 @@
|
||||
use std::time::Duration;
|
||||
|
||||
use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
|
||||
use crank_core::{CachedHeader, CachedResponse, HttpMethod, Target};
|
||||
use serde_json::{Value, json};
|
||||
use sha2::{Digest, Sha256};
|
||||
|
||||
use crate::{AdapterResponse, PreparedRequest, RuntimeOperation, RuntimeRequestContext};
|
||||
|
||||
pub(crate) fn response_cache_ttl(operation: &RuntimeOperation) -> Option<Duration> {
|
||||
let is_cacheable_protocol =
|
||||
matches!(&operation.target, Target::Rest(target) if target.method == HttpMethod::Get);
|
||||
if !is_cacheable_protocol {
|
||||
return None;
|
||||
}
|
||||
|
||||
let policy = operation.execution_config.response_cache.as_ref()?;
|
||||
if operation.execution_config.auth_profile_ref.is_some() || policy.ttl_ms == 0 {
|
||||
return None;
|
||||
}
|
||||
|
||||
Some(Duration::from_millis(policy.ttl_ms))
|
||||
}
|
||||
|
||||
pub(crate) fn response_cache_key(
|
||||
operation: &RuntimeOperation,
|
||||
prepared_request: &PreparedRequest,
|
||||
request_context: Option<&RuntimeRequestContext>,
|
||||
) -> Option<String> {
|
||||
let _ = response_cache_ttl(operation)?;
|
||||
let scope = request_context?.response_cache_scope()?;
|
||||
let fingerprint = json!({
|
||||
"path_params": prepared_request.path_params,
|
||||
"query_params": prepared_request.query_params,
|
||||
"headers": prepared_request.headers,
|
||||
"body": prepared_request.body,
|
||||
});
|
||||
let fingerprint_bytes = serde_json::to_vec(&fingerprint).ok()?;
|
||||
let fingerprint_hash = URL_SAFE_NO_PAD.encode(Sha256::digest(fingerprint_bytes));
|
||||
|
||||
Some(format!(
|
||||
"crank:response:workspace:{}:agent:{}:operation:{}:version:{}:request:{}",
|
||||
scope.workspace_key,
|
||||
scope.agent_key,
|
||||
operation.operation_id.as_str(),
|
||||
operation.operation_version,
|
||||
fingerprint_hash
|
||||
))
|
||||
}
|
||||
|
||||
pub(crate) fn cached_response_from_adapter(
|
||||
adapter_response: &AdapterResponse,
|
||||
) -> Option<CachedResponse> {
|
||||
let body = serde_json::to_vec(&adapter_response.body).ok()?;
|
||||
Some(CachedResponse {
|
||||
status: adapter_response.status_code,
|
||||
headers: adapter_response
|
||||
.headers
|
||||
.iter()
|
||||
.map(|(name, value)| CachedHeader {
|
||||
name: name.clone(),
|
||||
value: value.clone(),
|
||||
})
|
||||
.collect(),
|
||||
body,
|
||||
data: serde_json::to_vec(&adapter_response.data).ok()?,
|
||||
})
|
||||
}
|
||||
|
||||
pub(crate) fn adapter_response_from_cached(
|
||||
cached_response: CachedResponse,
|
||||
) -> Result<AdapterResponse, String> {
|
||||
let body: Value =
|
||||
serde_json::from_slice(&cached_response.body).map_err(|error| error.to_string())?;
|
||||
let data: Value =
|
||||
serde_json::from_slice(&cached_response.data).map_err(|error| error.to_string())?;
|
||||
Ok(AdapterResponse {
|
||||
status_code: cached_response.status,
|
||||
headers: cached_response
|
||||
.headers
|
||||
.into_iter()
|
||||
.map(|header| (header.name, header.value))
|
||||
.collect(),
|
||||
body,
|
||||
data,
|
||||
})
|
||||
}
|
||||
Reference in New Issue
Block a user