bsodfather/crank
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

runtime: switch executor to adapter registry

Browse Source
This commit is contained in:
github-ops
2026-05-14 18:28:19 +00:00
parent d9065a737d
commit 8ddcb213a3
12 changed files with 226 additions and 334 deletions
Download Patch File Download Diff File Expand all files Collapse all files
TASKS.md
+1
View File
@@ -151,5 +151,6 @@ Progress:
- Phase 0 / task `0.7`: введены `CapabilityProfile` и `CommunityCapabilityProfile`, а `admin-api` capability payload теперь собирается через public seam вместо локального literal
- Phase 0 / task `0.8`: введены `ProtocolAdapter`, `ProtocolAdapterError`, `AdapterRegistry` и базовые transport-типы (`PreparedRequest`, `AdapterResponse`, `WindowExecutionResult`, `RuntimeRequestContext`) в `crank_core::ext::protocol`
- Phase 0 / task `0.9`: `crank-adapter-rest::RestAdapter` реализует новый `ProtocolAdapter` contract, а seam дополнен явным `Target` и window parameters для реального adapter wiring
- Phase 0 / task `0.10`: `RuntimeExecutor` переведен с hardcoded adapter fields на `AdapterRegistry`, `crank-runtime` больше не держит protocol feature flags, а общий `PreparedRequest` seam дополнен `timeout_ms` для корректного runtime dispatch
- Phase 0 / task `0.13`: введены `RegistryExtension`, `ExtensionMigration` и `apply_extension_migrations(...)` в `crank-registry` как отдельный public seam для additive private migrations
- backward-compatible alias `CommunityMachineCredentialVerifier` сохранен, поведение Community не изменено
apps/admin-api/src/error.rs
+1
View File
@@ -423,6 +423,7 @@ fn runtime_test_failure_code(error: &RuntimeError) -> &'static str {
RuntimeError::GraphqlAdapter(_) => "runtime_graphql_error",
RuntimeError::GrpcAdapter(_) => "runtime_grpc_error",
RuntimeError::RestAdapter(_) => "runtime_rest_error",
RuntimeError::ProtocolAdapter(_) => "runtime_adapter_error",
RuntimeError::SoapAdapter(_) => "runtime_soap_error",
RuntimeError::WebsocketAdapter(_) => "runtime_websocket_error",
RuntimeError::UnsupportedProtocol { .. } => "runtime_protocol_error",
apps/admin-api/src/service.rs
+1
View File
@@ -4031,6 +4031,7 @@ fn runtime_error_code(error: &RuntimeError) -> &'static str {
RuntimeError::GraphqlAdapter(_) => "graphql_error",
RuntimeError::GrpcAdapter(_) => "grpc_error",
RuntimeError::RestAdapter(_) => "rest_error",
RuntimeError::ProtocolAdapter(_) => "adapter_error",
RuntimeError::SoapAdapter(_) => "soap_error",
RuntimeError::WebsocketAdapter(_) => "websocket_error",
RuntimeError::UnsupportedProtocol { .. } => "unsupported_protocol",
apps/mcp-server/src/app.rs
+1
View File
@@ -2378,6 +2378,7 @@ fn runtime_error_code(error: &RuntimeError) -> &'static str {
RuntimeError::GraphqlAdapter(_) => "adapter_execution_error",
RuntimeError::GrpcAdapter(_) => "adapter_execution_error",
RuntimeError::RestAdapter(_) => "adapter_execution_error",
RuntimeError::ProtocolAdapter(_) => "adapter_execution_error",
RuntimeError::SoapAdapter(_) => "adapter_execution_error",
RuntimeError::WebsocketAdapter(_) => "adapter_execution_error",
RuntimeError::UnsupportedProtocol { .. } => "unsupported_protocol",
crates/crank-adapter-rest/src/lib.rs
+2 -2
View File
@@ -36,7 +36,7 @@ impl ProtocolAdapter for RestAdapter {
query_params: prepared.query_params.clone(),
headers: prepared.headers.clone(),
body: prepared.body.clone(),
timeout_ms: 30_000,
timeout_ms: prepared.timeout_ms,
};
let response = self.execute(target, &request).await?;
@@ -63,7 +63,7 @@ impl ProtocolAdapter for RestAdapter {
query_params: prepared.query_params.clone(),
headers: prepared.headers.clone(),
body: prepared.body.clone(),
timeout_ms: 30_000,
timeout_ms: prepared.timeout_ms,
},
window_duration_ms,
max_items: max_items.map(|value| value.saturating_add(1)),
crates/crank-core/src/ext/protocol.rs
+2
View File
@@ -71,6 +71,8 @@ pub struct PreparedRequest {
pub variables: Option<Value>,
#[serde(skip_serializing_if = "Option::is_none")]
pub body: Option<Value>,
#[serde(default)]
pub timeout_ms: u64,
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
crates/crank-runtime/Cargo.toml
-4
View File
@@ -10,10 +10,6 @@ path = "src/lib.rs"
[features]
default = []
protocol-graphql = []
protocol-grpc = []
protocol-soap = []
protocol-websocket = []
[dependencies]
aes-gcm.workspace = true
crates/crank-runtime/src/error.rs
+2 -37
View File
@@ -4,15 +4,6 @@ use crank_mapping::MappingError;
use crank_schema::SchemaError;
use thiserror::Error;
#[cfg(feature = "protocol-graphql")]
use crank_adapter_graphql::GraphqlAdapterError;
#[cfg(feature = "protocol-grpc")]
use crank_adapter_grpc::GrpcAdapterError;
#[cfg(feature = "protocol-soap")]
use crank_adapter_soap::SoapAdapterError;
#[cfg(feature = "protocol-websocket")]
use crank_adapter_websocket::WebsocketAdapterError;
#[derive(Debug, Error)]
pub enum RuntimeError {
#[error(transparent)]
@@ -26,6 +17,8 @@ pub enum RuntimeError {
#[error(transparent)]
RestAdapter(#[from] RestAdapterError),
#[error("{0}")]
ProtocolAdapter(String),
#[error("{0}")]
SoapAdapter(String),
#[error("{0}")]
WebsocketAdapter(String),
@@ -58,31 +51,3 @@ pub enum RuntimeError {
details: String,
},
}
#[cfg(feature = "protocol-graphql")]
impl From<GraphqlAdapterError> for RuntimeError {
fn from(value: GraphqlAdapterError) -> Self {
Self::GraphqlAdapter(value.to_string())
}
}
#[cfg(feature = "protocol-grpc")]
impl From<GrpcAdapterError> for RuntimeError {
fn from(value: GrpcAdapterError) -> Self {
Self::GrpcAdapter(value.to_string())
}
}
#[cfg(feature = "protocol-soap")]
impl From<SoapAdapterError> for RuntimeError {
fn from(value: SoapAdapterError) -> Self {
Self::SoapAdapter(value.to_string())
}
}
#[cfg(feature = "protocol-websocket")]
impl From<WebsocketAdapterError> for RuntimeError {
fn from(value: WebsocketAdapterError) -> Self {
Self::WebsocketAdapter(value.to_string())
}
}
crates/crank-runtime/src/executor.rs
+134 -291
View File
@@ -3,10 +3,10 @@ use std::sync::Arc;
use std::time::Duration;
use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
use crank_adapter_rest::{RestAdapter, RestRequest, RestWindowRequest};
use crank_adapter_rest::RestAdapter;
use crank_core::{
CachedHeader, CachedResponse, ExecutionMode, HttpMethod, ResponseCacheStore, Target,
TransportBehavior,
AdapterRegistry, CachedHeader, CachedResponse, ExecutionMode, HttpMethod, ResponseCacheStore,
SharedProtocolAdapter, Target, TransportBehavior,
};
use serde_json::{Map, Value, json};
use sha2::{Digest, Sha256};
@@ -18,26 +18,9 @@ use crate::{
RuntimeRequestContext, WindowExecutionResult,
};
#[cfg(feature = "protocol-graphql")]
use crank_adapter_graphql::{GraphqlAdapter, GraphqlRequest};
#[cfg(feature = "protocol-grpc")]
use crank_adapter_grpc::{GrpcAdapter, GrpcRequest, GrpcWindowRequest};
#[cfg(feature = "protocol-soap")]
use crank_adapter_soap::{SoapAdapter, SoapRequest};
#[cfg(feature = "protocol-websocket")]
use crank_adapter_websocket::{WebsocketAdapter, WebsocketWindowRequest};
#[derive(Clone)]
pub struct RuntimeExecutor {
#[cfg(feature = "protocol-graphql")]
graphql_adapter: GraphqlAdapter,
#[cfg(feature = "protocol-grpc")]
grpc_adapter: GrpcAdapter,
rest_adapter: RestAdapter,
#[cfg(feature = "protocol-soap")]
soap_adapter: SoapAdapter,
#[cfg(feature = "protocol-websocket")]
websocket_adapter: WebsocketAdapter,
adapters: AdapterRegistry,
limits: RuntimeLimits,
unary_limiter: Arc<Semaphore>,
window_limiter: Arc<Semaphore>,
@@ -59,15 +42,7 @@ impl RuntimeExecutor {
pub fn with_limits(limits: RuntimeLimits) -> Self {
Self {
#[cfg(feature = "protocol-graphql")]
graphql_adapter: GraphqlAdapter::new(),
#[cfg(feature = "protocol-grpc")]
grpc_adapter: GrpcAdapter::new(),
rest_adapter: RestAdapter::new(),
#[cfg(feature = "protocol-soap")]
soap_adapter: SoapAdapter::new(),
#[cfg(feature = "protocol-websocket")]
websocket_adapter: WebsocketAdapter::new(),
adapters: community_adapter_registry(),
unary_limiter: Arc::new(Semaphore::new(limits.max_concurrent_unary)),
window_limiter: Arc::new(Semaphore::new(limits.max_concurrent_window)),
session_limiter: Arc::new(Semaphore::new(limits.max_concurrent_sessions)),
@@ -185,9 +160,6 @@ impl RuntimeExecutor {
let adapter_response = if matches!(
streaming.transport_behavior,
TransportBehavior::ServerStream
) && matches!(
operation.target,
Target::Rest(_) | Target::Grpc(_) | Target::Websocket(_)
) {
self.execute_window_adapter(operation, prepared_request, request_context)
.await?
@@ -322,131 +294,31 @@ impl RuntimeExecutor {
request_context: Option<&RuntimeRequestContext>,
) -> Result<AdapterResponse, RuntimeError> {
log_runtime_event("adapter.dispatch", operation, request_context);
let context_headers = runtime_context_headers(request_context);
match &operation.target {
Target::Grpc(target) => {
#[cfg(not(feature = "protocol-grpc"))]
{
let _ = target;
return Err(RuntimeError::UnsupportedProtocol {
protocol: operation.protocol,
});
}
#[cfg(feature = "protocol-grpc")]
{
let request = GrpcRequest {
headers: merge_headers(
&BTreeMap::new(),
&operation.execution_config.headers,
&prepared_request.headers,
&context_headers,
),
body: prepared_request
.grpc
.clone()
.unwrap_or(Value::Object(Map::new())),
timeout_ms: operation.execution_config.timeout_ms,
};
let response = self.grpc_adapter.execute(target, &request).await?;
Ok(AdapterResponse {
status_code: response.status_code,
headers: response.headers,
body: response.body.clone(),
data: response.body,
})
}
}
Target::Graphql(target) => {
#[cfg(not(feature = "protocol-graphql"))]
{
let _ = target;
return Err(RuntimeError::UnsupportedProtocol {
protocol: operation.protocol,
});
}
#[cfg(feature = "protocol-graphql")]
{
let request = GraphqlRequest {
headers: merge_headers(
&BTreeMap::new(),
&operation.execution_config.headers,
&prepared_request.headers,
&context_headers,
),
variables: prepared_request.variables.clone(),
timeout_ms: operation.execution_config.timeout_ms,
};
let response = self.graphql_adapter.execute(target, &request).await?;
Ok(AdapterResponse {
status_code: response.status_code,
headers: response.headers,
body: response.body,
data: response.data,
})
}
}
Target::Rest(target) => {
let request = RestRequest {
path_params: prepared_request.path_params.clone(),
query_params: prepared_request.query_params.clone(),
headers: merge_headers(
&target.static_headers,
&operation.execution_config.headers,
&prepared_request.headers,
&context_headers,
),
body: prepared_request.body.clone(),
timeout_ms: operation.execution_config.timeout_ms,
};
let response = self.rest_adapter.execute(target, &request).await?;
Ok(AdapterResponse {
status_code: response.status_code,
headers: response.headers,
body: response.body,
data: Value::Null,
})
}
Target::Soap(target) => {
#[cfg(not(feature = "protocol-soap"))]
{
let _ = target;
Err(RuntimeError::UnsupportedProtocol {
protocol: operation.protocol,
})
}
#[cfg(feature = "protocol-soap")]
{
let request = SoapRequest {
headers: merge_headers(
&BTreeMap::new(),
&operation.execution_config.headers,
&prepared_request.headers,
&context_headers,
),
body: prepared_request
.body
.clone()
.unwrap_or(Value::Object(Map::new())),
timeout_ms: operation.execution_config.timeout_ms,
};
let response = self.soap_adapter.execute(target, &request).await?;
Ok(AdapterResponse {
status_code: response.status_code,
headers: response.headers,
body: response.body,
data: Value::Null,
})
}
}
Target::Websocket(_) => Err(RuntimeError::UnsupportedExecutionMode {
let adapter = self.adapter_for(operation)?;
if !adapter.supports_mode(ExecutionMode::Unary) {
return Err(RuntimeError::UnsupportedExecutionMode {
operation_id: operation.operation_id.as_str().to_owned(),
mode: ExecutionMode::Unary,
}),
});
}
let prepared_request = adapter_prepared_request(
operation,
&prepared_request,
request_context,
operation.execution_config.timeout_ms,
);
let adapter_context = adapter_request_context(request_context);
adapter
.invoke_unary(
&operation.target,
&prepared_request.into(),
&adapter_context,
)
.await
.map(Into::into)
.map_err(|error| map_protocol_adapter_error(operation, error))
}
async fn execute_window_adapter(
@@ -456,145 +328,50 @@ impl RuntimeExecutor {
request_context: Option<&RuntimeRequestContext>,
) -> Result<AdapterResponse, RuntimeError> {
log_runtime_event("window.adapter.dispatch", operation, request_context);
let context_headers = runtime_context_headers(request_context);
match &operation.target {
Target::Rest(target) => {
let Some(streaming) = operation.execution_config.streaming.as_ref() else {
return Err(RuntimeError::MissingStreamingConfig {
operation_id: operation.operation_id.as_str().to_owned(),
});
};
let request = RestWindowRequest {
request: RestRequest {
path_params: prepared_request.path_params.clone(),
query_params: prepared_request.query_params.clone(),
headers: merge_headers(
&target.static_headers,
&operation.execution_config.headers,
&prepared_request.headers,
&context_headers,
),
body: prepared_request.body.clone(),
timeout_ms: streaming
.upstream_timeout_ms
.unwrap_or(operation.execution_config.timeout_ms),
},
window_duration_ms: streaming.window_duration_ms.unwrap_or_default(),
max_items: streaming.max_items.map(|value| value.saturating_add(1)),
};
let response = self.rest_adapter.execute_window(target, &request).await?;
Ok(AdapterResponse {
status_code: response.status_code,
headers: response.headers,
body: response.body,
data: Value::Null,
})
}
Target::Grpc(target) => {
#[cfg(not(feature = "protocol-grpc"))]
{
let _ = target;
return Err(RuntimeError::UnsupportedProtocol {
protocol: operation.protocol,
});
}
#[cfg(feature = "protocol-grpc")]
{
let Some(streaming) = operation.execution_config.streaming.as_ref() else {
return Err(RuntimeError::MissingStreamingConfig {
operation_id: operation.operation_id.as_str().to_owned(),
});
};
let request = GrpcWindowRequest {
request: GrpcRequest {
headers: merge_headers(
&BTreeMap::new(),
&operation.execution_config.headers,
&prepared_request.headers,
&context_headers,
),
body: prepared_request
.grpc
.clone()
.unwrap_or(Value::Object(Map::new())),
timeout_ms: streaming
.upstream_timeout_ms
.unwrap_or(operation.execution_config.timeout_ms),
},
window_duration_ms: streaming.window_duration_ms.unwrap_or_default(),
max_items: streaming.max_items.map(|value| value.saturating_add(1)),
};
let response = self.grpc_adapter.execute_window(target, &request).await?;
Ok(AdapterResponse {
status_code: response.status_code,
headers: response.headers,
body: response.body,
data: Value::Null,
})
}
}
Target::Websocket(target) => {
#[cfg(not(feature = "protocol-websocket"))]
{
let _ = target;
Err(RuntimeError::UnsupportedProtocol {
protocol: operation.protocol,
})
}
#[cfg(feature = "protocol-websocket")]
{
let Some(streaming) = operation.execution_config.streaming.as_ref() else {
return Err(RuntimeError::MissingStreamingConfig {
operation_id: operation.operation_id.as_str().to_owned(),
});
};
let websocket_options = operation
.execution_config
.protocol_options
.as_ref()
.and_then(|options| options.websocket.as_ref());
let request = WebsocketWindowRequest {
headers: merge_headers(
&BTreeMap::new(),
&operation.execution_config.headers,
&prepared_request.headers,
&context_headers,
),
window_duration_ms: streaming.window_duration_ms.unwrap_or_default(),
max_items: streaming.max_items.map(|value| value.saturating_add(1)),
heartbeat_interval_ms: websocket_options
.and_then(|options| options.heartbeat_interval_ms),
reconnect_max_attempts: websocket_options
.and_then(|options| options.reconnect_max_attempts)
.unwrap_or_default(),
reconnect_backoff_ms: websocket_options
.and_then(|options| options.reconnect_backoff_ms)
.unwrap_or_default(),
};
let response = self
.websocket_adapter
.execute_window(target, &request)
.await?;
Ok(AdapterResponse {
status_code: response.status_code,
headers: response.headers,
body: response.body,
data: Value::Null,
})
}
}
Target::Soap(_) => Err(RuntimeError::UnsupportedExecutionMode {
let Some(streaming) = operation.execution_config.streaming.as_ref() else {
return Err(RuntimeError::MissingStreamingConfig {
operation_id: operation.operation_id.as_str().to_owned(),
});
};
let adapter = self.adapter_for(operation)?;
if !adapter.supports_mode(ExecutionMode::Window) {
return Err(RuntimeError::UnsupportedExecutionMode {
operation_id: operation.operation_id.as_str().to_owned(),
mode: ExecutionMode::Window,
}),
_ => {
self.execute_adapter(operation, prepared_request, request_context)
.await
}
});
}
let prepared_request = adapter_prepared_request(
operation,
&prepared_request,
request_context,
streaming
.upstream_timeout_ms
.unwrap_or(operation.execution_config.timeout_ms),
);
let adapter_context = adapter_request_context(request_context);
let result = adapter
.invoke_window(
&operation.target,
&prepared_request.into(),
streaming.window_duration_ms.unwrap_or_default(),
streaming.max_items,
&adapter_context,
)
.await
.map_err(|error| map_protocol_adapter_error(operation, error))?;
Ok(AdapterResponse {
status_code: 200,
headers: BTreeMap::new(),
body: json!({
"summary": result.summary,
"items": result.items,
"cursor": result.cursor,
"done": result.window_complete,
}),
data: Value::Null,
})
}
fn acquire_unary_permit(
@@ -732,10 +509,76 @@ impl PreparedRequest {
grpc: non_empty_payload(request.get("grpc").cloned()),
variables: non_empty_payload(request.get("variables").cloned()),
body: request.get("body").and_then(non_empty_body).cloned(),
timeout_ms: 0,
})
}
}
fn community_adapter_registry() -> AdapterRegistry {
AdapterRegistry::empty().register(Arc::new(RestAdapter::new()) as SharedProtocolAdapter)
}
fn adapter_request_context(
request_context: Option<&RuntimeRequestContext>,
) -> crank_core::RuntimeRequestContext {
request_context
.map(Into::into)
.unwrap_or_else(|| crank_core::RuntimeRequestContext::new(String::new(), String::new()))
}
fn adapter_prepared_request(
operation: &RuntimeOperation,
prepared_request: &PreparedRequest,
request_context: Option<&RuntimeRequestContext>,
timeout_ms: u64,
) -> PreparedRequest {
let empty_headers = BTreeMap::new();
let static_headers = match &operation.target {
Target::Rest(target) => &target.static_headers,
_ => &empty_headers,
};
let mut prepared_request = prepared_request.clone();
prepared_request.headers = merge_headers(
static_headers,
&operation.execution_config.headers,
&prepared_request.headers,
&runtime_context_headers(request_context),
);
prepared_request.timeout_ms = timeout_ms;
prepared_request
}
fn map_protocol_adapter_error(
operation: &RuntimeOperation,
error: crank_core::ProtocolAdapterError,
) -> RuntimeError {
match error {
crank_core::ProtocolAdapterError::UnsupportedMode { mode, .. } => {
RuntimeError::UnsupportedExecutionMode {
operation_id: operation.operation_id.as_str().to_owned(),
mode,
}
}
crank_core::ProtocolAdapterError::Message(message) => RuntimeError::ProtocolAdapter(
format!("operation {}: {message}", operation.operation_id),
),
}
}
impl RuntimeExecutor {
fn adapter_for(
&self,
operation: &RuntimeOperation,
) -> Result<SharedProtocolAdapter, RuntimeError> {
self.adapters
.get(operation.protocol)
.ok_or(RuntimeError::UnsupportedProtocol {
protocol: operation.protocol,
})
}
}
fn finalize_output(
operation: &RuntimeOperation,
response: &AdapterResponse,
crates/crank-runtime/src/model.rs
+41
View File
@@ -53,6 +53,8 @@ pub struct PreparedRequest {
pub variables: Option<Value>,
#[serde(skip_serializing_if = "Option::is_none")]
pub body: Option<Value>,
#[serde(default)]
pub timeout_ms: u64,
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
@@ -63,3 +65,42 @@ pub struct AdapterResponse {
pub body: Value,
pub data: Value,
}
impl From<PreparedRequest> for crank_core::PreparedRequest {
fn from(value: PreparedRequest) -> Self {
Self {
path_params: value.path_params,
query_params: value.query_params,
headers: value.headers,
grpc: value.grpc,
variables: value.variables,
body: value.body,
timeout_ms: value.timeout_ms,
}
}
}
impl From<crank_core::PreparedRequest> for PreparedRequest {
fn from(value: crank_core::PreparedRequest) -> Self {
Self {
path_params: value.path_params,
query_params: value.query_params,
headers: value.headers,
grpc: value.grpc,
variables: value.variables,
body: value.body,
timeout_ms: value.timeout_ms,
}
}
}
impl From<crank_core::AdapterResponse> for AdapterResponse {
fn from(value: crank_core::AdapterResponse) -> Self {
Self {
status_code: value.status_code,
headers: value.headers,
body: value.body,
data: value.data,
}
}
}
crates/crank-runtime/src/request_context.rs
+28
View File
@@ -51,6 +51,34 @@ impl RuntimeRequestContext {
}
}
impl From<ResponseCacheScope> for crank_core::ResponseCacheScope {
fn from(value: ResponseCacheScope) -> Self {
Self {
workspace_key: value.workspace_key,
agent_key: value.agent_key,
}
}
}
impl From<&ResponseCacheScope> for crank_core::ResponseCacheScope {
fn from(value: &ResponseCacheScope) -> Self {
Self {
workspace_key: value.workspace_key.clone(),
agent_key: value.agent_key.clone(),
}
}
}
impl From<&RuntimeRequestContext> for crank_core::RuntimeRequestContext {
fn from(value: &RuntimeRequestContext) -> Self {
Self {
request_id: value.request_id.clone(),
correlation_id: value.correlation_id.clone(),
response_cache_scope: value.response_cache_scope.as_ref().map(Into::into),
}
}
}
#[cfg(test)]
mod tests {
use super::RuntimeRequestContext;
crates/crank-runtime/src/streaming.rs
+13
View File
@@ -11,3 +11,16 @@ pub struct WindowExecutionResult {
pub truncated: bool,
pub has_more: bool,
}
impl From<crank_core::WindowExecutionResult> for WindowExecutionResult {
fn from(value: crank_core::WindowExecutionResult) -> Self {
Self {
summary: value.summary,
items: value.items,
cursor: value.cursor,
window_complete: value.window_complete,
truncated: value.truncated,
has_more: value.has_more,
}
}
}