feat(adapters): from_wss consumer adapter behind the wss feature (ADR-070)
- FromWss: dial wss:// -> split_tungstenite_to_bytes (client-side twin of the axum WS byte-adapter; one seam, both directions, OQ-01) -> Connection::from_bidi(b"alk/channels") -> alkcall ChannelClient (channel 0 install + dispatch loop) -> alkcall from_call importer. No protocol fork: specs mirror the remote, provenance FromCall. - Drop semantics (OQ-03 v1): session drop -> monitor fails all in-flight pendings retryable CONNECTION_CLOSED (WsPumps::read_eof Notify); no 30s-deadline hang. - Bearer token via constructor/assembly layer (ADR-014 no-env-vars). Production fix in the WS server half (upgrade.rs): the upgrade identity now propagates to channel 0's CallConnection (was AuthContext::anonymous -> dispatcher saw no identity, ACL checks ran unauthenticated; services/list filtered scoped ops for all callers). 9 in-module tests incl. full round-trip consumer<->server (both halves of the adapter together), ACL end-to-end, drop-no-hang. Verified: cargo test (227 lib default), --all-features (227 lib + 5 MCP + 10 WS integration), clippy -D warnings (both), fmt.
This commit is contained in:
@@ -0,0 +1,624 @@
|
||||
//! `from_wss` consumer adapter ([ADR-070]): connect to a remote node's WSS
|
||||
//! endpoint, run the channels-over-WS session (the consumer half), and
|
||||
//! import the remote node's operations as forwarding handlers — the
|
||||
//! same-protocol importer (`from_call` pattern), with WSS as the transport.
|
||||
//!
|
||||
//! Feature-gated behind `wss` (tokio-tungstenite). The dial carries
|
||||
//! `Authorization: Bearer <token>` when constructed with an auth token; at
|
||||
//! call time the imported handlers read the per-call credential from
|
||||
//! `OperationContext.capabilities` (the no-env-vars path, ADR-014), never
|
||||
//! from `std::env::var`. Provenance is `FromCall` (leaf,
|
||||
//! `composition_authority: None`, `scoped_env: None`, `Internal` by
|
||||
//! default — ADR-015/022), because the imported session IS the call
|
||||
//! protocol.
|
||||
//!
|
||||
//! Import flow (ADR-070): dial WSS → adapt the tungstenite stream with the
|
||||
//! shared WS↔byte-stream seam
|
||||
//! ([`crate::websocket::split_tungstenite_to_bytes`]) →
|
||||
//! `Connection::from_bidi(_, b"alk/channels")` → alkcall `ChannelClient`
|
||||
//! (channel 0 install + client dispatch loop) → `services/list` +
|
||||
//! `services/schema` over channel 0 → one forwarding `HandlerRegistration`
|
||||
//! per discovered op via alkcall's `from_call` importer.
|
||||
//!
|
||||
//! Reconnect policy (OQ-03 disposition): v1 = none. A connection drop
|
||||
//! fails in-flight calls retryable: alkcall's client-side read pump only
|
||||
//! routes envelopes and does not observe EOF, so the adapter owns drop
|
||||
//! semantics — the [`WssSession`] monitor awaits the WS read pump and
|
||||
//! fails all pending calls with retryable `CONNECTION_CLOSED`. Subsequent
|
||||
//! handler calls fail on write; reconnect policy is the assembly layer's
|
||||
//! job.
|
||||
//!
|
||||
//! [ADR-070]: crate::docs
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use alkcall::channels::client::ChannelClient;
|
||||
use alkcall::client::{
|
||||
from_call as import_from_call, AdapterError, FromCallConfig, OperationAdapter,
|
||||
};
|
||||
use alkcall::core::types::Connection;
|
||||
use alkcall::protocol::connection::CallConnection;
|
||||
use alkcall::protocol::wire::CallError;
|
||||
use tokio_tungstenite::tungstenite::client::IntoClientRequest;
|
||||
|
||||
use crate::websocket::split_tungstenite_to_bytes;
|
||||
|
||||
pub struct FromWss {
|
||||
endpoint: String,
|
||||
auth_token: Option<String>,
|
||||
namespace: Option<String>,
|
||||
}
|
||||
|
||||
impl FromWss {
|
||||
pub fn new(endpoint: impl Into<String>) -> Self {
|
||||
Self {
|
||||
endpoint: endpoint.into(),
|
||||
auth_token: None,
|
||||
namespace: None,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn with_auth_token(mut self, token: impl Into<String>) -> Self {
|
||||
self.auth_token = Some(token.into());
|
||||
self
|
||||
}
|
||||
|
||||
pub fn with_namespace(mut self, namespace: impl Into<String>) -> Self {
|
||||
self.namespace = Some(namespace.into());
|
||||
self
|
||||
}
|
||||
|
||||
pub fn endpoint(&self) -> &str {
|
||||
&self.endpoint
|
||||
}
|
||||
|
||||
pub fn namespace(&self) -> Option<&str> {
|
||||
self.namespace.as_deref()
|
||||
}
|
||||
|
||||
pub fn auth_token(&self) -> Option<&str> {
|
||||
self.auth_token.as_deref()
|
||||
}
|
||||
}
|
||||
|
||||
/// A live `from_wss` session: the channel-0 `CallConnection` (held
|
||||
/// exclusively by the consumer) plus the drop-monitor tasks. Dropping
|
||||
/// the session (via [`WssSession::drop`] or `std::mem::forget` for
|
||||
/// fire-and-forget imports) tears the WS down; the monitor fails all
|
||||
/// in-flight pending calls with retryable `CONNECTION_CLOSED`.
|
||||
pub struct WssSession {
|
||||
_client: ChannelClient,
|
||||
pub call_connection: Arc<CallConnection>,
|
||||
_monitor: WssDropMonitor,
|
||||
}
|
||||
|
||||
/// The drop monitor: fires `fail_all` on WS read EOF or on explicit
|
||||
/// session close (the session's `Drop` impl sends the close signal).
|
||||
struct WssDropMonitor {
|
||||
close_tx: Option<tokio::sync::oneshot::Sender<()>>,
|
||||
}
|
||||
|
||||
impl Drop for WssDropMonitor {
|
||||
fn drop(&mut self) {
|
||||
if let Some(tx) = self.close_tx.take() {
|
||||
let _ = tx.send(());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl WssSession {
|
||||
/// Dial the WSS endpoint and establish the channels consumer session.
|
||||
/// Exposed for the assembly layer and tests: hold the session for as
|
||||
/// long as imported ops should stay callable; on drop, in-flight
|
||||
/// calls fail retryable (no hang until the 30s sweeper deadline).
|
||||
pub async fn connect(endpoint: &str, auth_token: Option<&str>) -> Result<Self, AdapterError> {
|
||||
let mut request = endpoint
|
||||
.into_client_request()
|
||||
.map_err(|e| AdapterError::Transport {
|
||||
message: format!("invalid WSS endpoint `{endpoint}`: {e}"),
|
||||
})?;
|
||||
if let Some(token) = auth_token {
|
||||
let value = format!("Bearer {token}");
|
||||
request.headers_mut().insert(
|
||||
"Authorization",
|
||||
value.parse().map_err(|_| AdapterError::Transport {
|
||||
message: "bearer token is not a valid header value".to_string(),
|
||||
})?,
|
||||
);
|
||||
}
|
||||
|
||||
let (ws, _response) = tokio_tungstenite::connect_async(request)
|
||||
.await
|
||||
.map_err(|e| AdapterError::Transport {
|
||||
message: format!("WSS connect failed: {e}"),
|
||||
})?;
|
||||
|
||||
let (byte_stream, pumps) = split_tungstenite_to_bytes(ws);
|
||||
let conn = Connection::from_bidi(byte_stream, b"alk/channels".to_vec(), None);
|
||||
|
||||
let client =
|
||||
ChannelClient::from_connection(conn)
|
||||
.await
|
||||
.map_err(|e| AdapterError::Transport {
|
||||
message: format!("channels session setup failed: {e}"),
|
||||
})?;
|
||||
let call_connection =
|
||||
client
|
||||
.take_call_connection()
|
||||
.await
|
||||
.ok_or_else(|| AdapterError::Transport {
|
||||
message: "channel client closed before channel 0 was installed".to_string(),
|
||||
})?;
|
||||
let call_connection = Arc::new(call_connection);
|
||||
|
||||
let (close_tx, close_rx) = tokio::sync::oneshot::channel();
|
||||
let pending = Arc::clone(call_connection.pending());
|
||||
tokio::spawn(async move {
|
||||
tokio::select! {
|
||||
_ = pumps.read_eof() => {},
|
||||
_ = close_rx => {},
|
||||
}
|
||||
pending.lock().fail_all(CallError::new(
|
||||
"CONNECTION_CLOSED",
|
||||
"from_wss connection dropped",
|
||||
true,
|
||||
));
|
||||
});
|
||||
|
||||
Ok(Self {
|
||||
_client: client,
|
||||
call_connection,
|
||||
_monitor: WssDropMonitor {
|
||||
close_tx: Some(close_tx),
|
||||
},
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl OperationAdapter for FromWss {
|
||||
async fn import(
|
||||
&self,
|
||||
) -> Result<Vec<alkcall::registry::registration::HandlerRegistration>, AdapterError> {
|
||||
let session = WssSession::connect(&self.endpoint, self.auth_token.as_deref()).await?;
|
||||
let config = match &self.namespace {
|
||||
Some(ns) => FromCallConfig::new().with_namespace_prefix(ns),
|
||||
None => FromCallConfig::new(),
|
||||
};
|
||||
let bundles = import_from_call(&session.call_connection, config).await;
|
||||
// Fire-and-forget: the imported handlers keep working off the
|
||||
// session's Arc'd CallConnection; the session's tasks are
|
||||
// detached (tokio semantics). Dropping the session here would
|
||||
// close the connection before the caller invokes anything.
|
||||
std::mem::forget(session);
|
||||
bundles
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use alkcall::core::auth::{Identity, IdentityProvider};
|
||||
use alkcall::protocol::wire::ResponseEnvelope;
|
||||
use alkcall::registry::discovery::{
|
||||
services_list_handler, services_list_spec, services_schema_handler, services_schema_spec,
|
||||
};
|
||||
use alkcall::registry::registration::{
|
||||
make_handler, HandlerKind, HandlerRegistration, OperationProvenance, OperationRegistry,
|
||||
};
|
||||
use alkcall::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Mutex as StdMutex;
|
||||
|
||||
fn identity(id: &str, scopes: &[&str]) -> Identity {
|
||||
Identity {
|
||||
id: id.to_string(),
|
||||
scopes: scopes.iter().map(|s| s.to_string()).collect(),
|
||||
resources: HashMap::new(),
|
||||
}
|
||||
}
|
||||
|
||||
struct StaticTokens {
|
||||
tokens: StdMutex<HashMap<String, Identity>>,
|
||||
}
|
||||
|
||||
impl IdentityProvider for StaticTokens {
|
||||
fn resolve_from_fingerprint(&self, _: &str) -> Option<Identity> {
|
||||
None
|
||||
}
|
||||
fn resolve_from_token(&self, token: &alkcall::core::auth::AuthToken) -> Option<Identity> {
|
||||
let s = String::from_utf8_lossy(&token.raw).to_string();
|
||||
self.tokens
|
||||
.lock()
|
||||
.unwrap_or_else(|e| e.into_inner())
|
||||
.get(&s)
|
||||
.cloned()
|
||||
}
|
||||
}
|
||||
|
||||
fn provider_with(tokens: Vec<(&str, Identity)>) -> Arc<dyn IdentityProvider> {
|
||||
let map: HashMap<String, Identity> = tokens
|
||||
.into_iter()
|
||||
.map(|(t, i)| (t.to_string(), i))
|
||||
.collect();
|
||||
Arc::new(StaticTokens {
|
||||
tokens: StdMutex::new(map),
|
||||
})
|
||||
}
|
||||
|
||||
fn echo_handler() -> alkcall::registry::registration::Handler {
|
||||
make_handler(|input, ctx| async move { ResponseEnvelope::ok(ctx.request_id, input) })
|
||||
}
|
||||
|
||||
fn noop_context(request_id: &str) -> alkcall::registry::context::OperationContext {
|
||||
struct NoopEnv;
|
||||
#[async_trait::async_trait]
|
||||
impl alkcall::registry::env::OperationEnv for NoopEnv {
|
||||
async fn invoke_with_policy(
|
||||
&self,
|
||||
_ns: &str,
|
||||
_op: &str,
|
||||
_input: serde_json::Value,
|
||||
parent: &alkcall::registry::context::OperationContext,
|
||||
_policy: alkcall::registry::context::AbortPolicy,
|
||||
) -> ResponseEnvelope {
|
||||
ResponseEnvelope::ok(parent.request_id.clone(), serde_json::Value::Null)
|
||||
}
|
||||
fn contains(&self, _name: &str) -> bool {
|
||||
false
|
||||
}
|
||||
}
|
||||
alkcall::registry::context::OperationContext {
|
||||
request_id: request_id.to_string(),
|
||||
parent_request_id: None,
|
||||
identity: None,
|
||||
handler_identity: None,
|
||||
forwarded_for: None,
|
||||
capabilities: alkcall::core::types::Capabilities::new(),
|
||||
metadata: HashMap::new(),
|
||||
scoped_env: alkcall::registry::context::ScopedPeerEnv::empty(),
|
||||
env: Arc::new(NoopEnv),
|
||||
abort_policy: alkcall::registry::context::AbortPolicy::default(),
|
||||
deadline: Some(std::time::Instant::now() + std::time::Duration::from_secs(30)),
|
||||
internal: true,
|
||||
ownership: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// A producer registry: `echo/run` (open), `admin/run` (admin scope),
|
||||
/// plus the discovery ops the importer calls over channel 0.
|
||||
fn producer_registry() -> Arc<OperationRegistry> {
|
||||
let mut inner = OperationRegistry::new();
|
||||
inner
|
||||
.register(HandlerRegistration::new(
|
||||
OperationSpec::new(
|
||||
"echo/run",
|
||||
OperationType::Query,
|
||||
Visibility::External,
|
||||
serde_json::json!({}),
|
||||
serde_json::json!({}),
|
||||
vec![],
|
||||
AccessControl::default(),
|
||||
None,
|
||||
),
|
||||
HandlerKind::Once(echo_handler()),
|
||||
OperationProvenance::Local,
|
||||
None,
|
||||
None,
|
||||
alkcall::core::types::Capabilities::new(),
|
||||
))
|
||||
.unwrap();
|
||||
inner
|
||||
.register(HandlerRegistration::new(
|
||||
OperationSpec::new(
|
||||
"admin/run",
|
||||
OperationType::Query,
|
||||
Visibility::External,
|
||||
serde_json::json!({}),
|
||||
serde_json::json!({}),
|
||||
vec![],
|
||||
AccessControl {
|
||||
required_scopes: vec!["admin".to_string()],
|
||||
..Default::default()
|
||||
},
|
||||
None,
|
||||
),
|
||||
HandlerKind::Once(echo_handler()),
|
||||
OperationProvenance::Local,
|
||||
None,
|
||||
None,
|
||||
alkcall::core::types::Capabilities::new(),
|
||||
))
|
||||
.unwrap();
|
||||
let inner = Arc::new(inner);
|
||||
|
||||
let mut registry = OperationRegistry::new();
|
||||
registry
|
||||
.register(HandlerRegistration::new(
|
||||
services_list_spec(),
|
||||
HandlerKind::Once(services_list_handler(Arc::clone(&inner))),
|
||||
OperationProvenance::Local,
|
||||
None,
|
||||
None,
|
||||
alkcall::core::types::Capabilities::new(),
|
||||
))
|
||||
.unwrap();
|
||||
registry
|
||||
.register(HandlerRegistration::new(
|
||||
services_schema_spec(),
|
||||
HandlerKind::Once(services_schema_handler(Arc::clone(&inner))),
|
||||
OperationProvenance::Local,
|
||||
None,
|
||||
None,
|
||||
alkcall::core::types::Capabilities::new(),
|
||||
))
|
||||
.unwrap();
|
||||
for spec in inner.list_operations() {
|
||||
let name = spec.name.clone();
|
||||
let reg = inner.registration(&name).unwrap();
|
||||
registry
|
||||
.register(HandlerRegistration::new(
|
||||
reg.spec.clone(),
|
||||
reg.handler.clone(),
|
||||
reg.provenance,
|
||||
reg.composition_authority.clone(),
|
||||
reg.scoped_env.clone(),
|
||||
reg.capabilities.clone(),
|
||||
))
|
||||
.unwrap();
|
||||
}
|
||||
Arc::new(registry)
|
||||
}
|
||||
|
||||
/// A producer registry with a never-responding `slow/op`.
|
||||
fn slow_producer_registry() -> Arc<OperationRegistry> {
|
||||
let mut inner = OperationRegistry::new();
|
||||
inner
|
||||
.register(HandlerRegistration::new(
|
||||
OperationSpec::new(
|
||||
"slow/op",
|
||||
OperationType::Query,
|
||||
Visibility::External,
|
||||
serde_json::json!({}),
|
||||
serde_json::json!({}),
|
||||
vec![],
|
||||
AccessControl::default(),
|
||||
None,
|
||||
),
|
||||
HandlerKind::Once(make_handler(|_input, _ctx| async move {
|
||||
tokio::time::sleep(std::time::Duration::from_secs(30)).await;
|
||||
ResponseEnvelope::ok("never", serde_json::json!({}))
|
||||
})),
|
||||
OperationProvenance::Local,
|
||||
None,
|
||||
None,
|
||||
alkcall::core::types::Capabilities::new(),
|
||||
))
|
||||
.unwrap();
|
||||
let inner = Arc::new(inner);
|
||||
let mut registry = OperationRegistry::new();
|
||||
registry
|
||||
.register(HandlerRegistration::new(
|
||||
services_list_spec(),
|
||||
HandlerKind::Once(services_list_handler(Arc::clone(&inner))),
|
||||
OperationProvenance::Local,
|
||||
None,
|
||||
None,
|
||||
alkcall::core::types::Capabilities::new(),
|
||||
))
|
||||
.unwrap();
|
||||
registry
|
||||
.register(HandlerRegistration::new(
|
||||
services_schema_spec(),
|
||||
HandlerKind::Once(services_schema_handler(Arc::clone(&inner))),
|
||||
OperationProvenance::Local,
|
||||
None,
|
||||
None,
|
||||
alkcall::core::types::Capabilities::new(),
|
||||
))
|
||||
.unwrap();
|
||||
for spec in inner.list_operations() {
|
||||
let name = spec.name.clone();
|
||||
let reg = inner.registration(&name).unwrap();
|
||||
registry
|
||||
.register(HandlerRegistration::new(
|
||||
reg.spec.clone(),
|
||||
reg.handler.clone(),
|
||||
reg.provenance,
|
||||
reg.composition_authority.clone(),
|
||||
reg.scoped_env.clone(),
|
||||
reg.capabilities.clone(),
|
||||
))
|
||||
.unwrap();
|
||||
}
|
||||
Arc::new(registry)
|
||||
}
|
||||
|
||||
async fn spawn_producer(
|
||||
registry: Arc<OperationRegistry>,
|
||||
provider: Arc<dyn IdentityProvider>,
|
||||
) -> String {
|
||||
let app = axum::Router::new()
|
||||
.route(
|
||||
"/alk/channels",
|
||||
axum::routing::get(crate::websocket::ws_upgrade_handler),
|
||||
)
|
||||
.layer(axum::middleware::from_fn_with_state(
|
||||
provider,
|
||||
crate::websocket::ws_bearer_auth,
|
||||
))
|
||||
.with_state(registry);
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let addr = format!("ws://{}", listener.local_addr().unwrap());
|
||||
tokio::spawn(async move {
|
||||
axum::serve(listener, app).await.unwrap();
|
||||
});
|
||||
format!("{addr}/alk/channels")
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn struct_holds_endpoint_token_namespace() {
|
||||
let adapter = FromWss::new("ws://localhost:9000/alk/channels");
|
||||
assert_eq!(adapter.endpoint(), "ws://localhost:9000/alk/channels");
|
||||
assert_eq!(adapter.namespace(), None);
|
||||
assert_eq!(adapter.auth_token(), None);
|
||||
|
||||
let with_all = adapter.with_auth_token("tok").with_namespace("remote");
|
||||
assert_eq!(with_all.auth_token(), Some("tok"));
|
||||
assert_eq!(with_all.namespace(), Some("remote"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn import_discovers_ops_and_builds_forwarding_handlers() {
|
||||
let endpoint = spawn_producer(
|
||||
producer_registry(),
|
||||
provider_with(vec![("tok-1", identity("alice", &["user", "admin"]))]),
|
||||
)
|
||||
.await;
|
||||
|
||||
let adapter = FromWss::new(&endpoint).with_auth_token("tok-1");
|
||||
let bundles = adapter.import().await.expect("import succeeds");
|
||||
let mut names: Vec<&str> = bundles.iter().map(|b| b.spec.name.as_str()).collect();
|
||||
names.sort();
|
||||
assert_eq!(names, vec!["admin/run", "echo/run"]);
|
||||
for b in &bundles {
|
||||
assert_eq!(b.provenance, OperationProvenance::FromCall);
|
||||
assert!(b.composition_authority.is_none());
|
||||
assert!(b.scoped_env.is_none());
|
||||
}
|
||||
// The spec mirrors the remote (all-External here; ADR-017 §3.
|
||||
// The assembly layer may override to Internal per ADR-015).
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn imported_ops_invoke_end_to_end() {
|
||||
let endpoint = spawn_producer(
|
||||
producer_registry(),
|
||||
provider_with(vec![("tok-1", identity("alice", &["user", "admin"]))]),
|
||||
)
|
||||
.await;
|
||||
|
||||
let adapter = FromWss::new(&endpoint).with_auth_token("tok-1");
|
||||
let bundles = adapter.import().await.expect("import succeeds");
|
||||
let echo = bundles
|
||||
.into_iter()
|
||||
.find(|b| b.spec.name == "echo/run")
|
||||
.expect("echo/run present");
|
||||
|
||||
let ctx = noop_context("req-e2e");
|
||||
let response = match &echo.handler {
|
||||
HandlerKind::Once(h) => h(serde_json::json!({ "hello": "world" }), ctx).await,
|
||||
HandlerKind::Stream(_) | HandlerKind::Sink(_) => {
|
||||
panic!("expected Once handler for query op")
|
||||
}
|
||||
};
|
||||
assert_eq!(response.request_id, "req-e2e");
|
||||
assert_eq!(response.result, Ok(serde_json::json!({ "hello": "world" })));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn acl_enforced_end_to_end() {
|
||||
// alice (no admin scope): services/list filters `admin/run` out
|
||||
// of discovery, so only echo/run is imported.
|
||||
let endpoint = spawn_producer(
|
||||
producer_registry(),
|
||||
provider_with(vec![("tok-alice", identity("alice", &["user"]))]),
|
||||
)
|
||||
.await;
|
||||
|
||||
let adapter = FromWss::new(&endpoint).with_auth_token("tok-alice");
|
||||
let bundles = adapter.import().await.expect("import succeeds");
|
||||
let names: Vec<&str> = bundles.iter().map(|b| b.spec.name.as_str()).collect();
|
||||
assert_eq!(names, vec!["echo/run"], "ACL-filtered op not discovered");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn namespace_prefix_applies_to_imported_names() {
|
||||
let endpoint = spawn_producer(
|
||||
producer_registry(),
|
||||
provider_with(vec![("tok-1", identity("alice", &["user", "admin"]))]),
|
||||
)
|
||||
.await;
|
||||
|
||||
let adapter = FromWss::new(&endpoint)
|
||||
.with_auth_token("tok-1")
|
||||
.with_namespace("remote");
|
||||
let bundles = adapter.import().await.expect("import succeeds");
|
||||
let mut names: Vec<&str> = bundles.iter().map(|b| b.spec.name.as_str()).collect();
|
||||
names.sort();
|
||||
assert_eq!(names, vec!["remote/admin/run", "remote/echo/run"]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn connection_drop_fails_in_flight_calls_retryable_no_hang() {
|
||||
let endpoint = spawn_producer(
|
||||
slow_producer_registry(),
|
||||
provider_with(vec![("tok-1", identity("alice", &[]))]),
|
||||
)
|
||||
.await;
|
||||
|
||||
let session = WssSession::connect(&endpoint, Some("tok-1"))
|
||||
.await
|
||||
.expect("connect");
|
||||
|
||||
let config = FromCallConfig::new();
|
||||
let bundles = import_from_call(&session.call_connection, config)
|
||||
.await
|
||||
.expect("import");
|
||||
let slow = bundles
|
||||
.into_iter()
|
||||
.find(|b| b.spec.name == "slow/op")
|
||||
.expect("slow/op present");
|
||||
|
||||
let ctx = noop_context("req-drop");
|
||||
let handler = match &slow.handler {
|
||||
HandlerKind::Once(h) => h.clone(),
|
||||
_ => panic!("expected Once handler"),
|
||||
};
|
||||
let call_task = tokio::spawn(async move { handler(serde_json::json!({}), ctx).await });
|
||||
|
||||
// Let the call go in-flight, then drop the session → WS close →
|
||||
// read EOF → the monitor's fail_all resolves the pending call.
|
||||
tokio::time::sleep(std::time::Duration::from_millis(200)).await;
|
||||
drop(session);
|
||||
|
||||
let response = tokio::time::timeout(std::time::Duration::from_secs(5), call_task)
|
||||
.await
|
||||
.expect("call resolves after drop (no hang until the 30s deadline)")
|
||||
.expect("join");
|
||||
match response.result {
|
||||
Err(e) => {
|
||||
assert!(e.retryable, "drop error must be retryable, got {e:?}");
|
||||
}
|
||||
Ok(_) => panic!("expected Err after connection drop"),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn unreachable_endpoint_returns_transport_error() {
|
||||
let adapter = FromWss::new("ws://127.0.0.1:1/alk/channels");
|
||||
match adapter.import().await {
|
||||
Ok(_) => panic!("expected Err for unreachable endpoint"),
|
||||
Err(AdapterError::Transport { .. }) => {}
|
||||
Err(other) => panic!("expected Transport, got {other}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn invalid_endpoint_uri_returns_transport_error() {
|
||||
let adapter = FromWss::new("not a url");
|
||||
match adapter.import().await {
|
||||
Ok(_) => panic!("expected Err for invalid endpoint"),
|
||||
Err(AdapterError::Transport { .. }) => {}
|
||||
Err(other) => panic!("expected Transport, got {other}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn no_env_vars_used_for_credentials() {
|
||||
std::env::set_var("WSS_TOKEN", "should-not-be-used");
|
||||
let adapter = FromWss::new("ws://localhost/alk/channels");
|
||||
assert!(adapter.auth_token().is_none());
|
||||
std::env::remove_var("WSS_TOKEN");
|
||||
}
|
||||
}
|
||||
@@ -11,6 +11,8 @@ pub mod to_openapi;
|
||||
|
||||
#[cfg(feature = "mcp")]
|
||||
pub mod from_mcp;
|
||||
#[cfg(feature = "wss")]
|
||||
pub mod from_wss;
|
||||
#[cfg(feature = "mcp")]
|
||||
pub mod to_mcp;
|
||||
|
||||
@@ -22,5 +24,7 @@ pub use to_openapi::to_openapi;
|
||||
|
||||
#[cfg(feature = "mcp")]
|
||||
pub use from_mcp::FromMCP;
|
||||
#[cfg(feature = "wss")]
|
||||
pub use from_wss::FromWss;
|
||||
#[cfg(feature = "mcp")]
|
||||
pub use to_mcp::{to_mcp_service, ToMcpGateway, ToMcpService};
|
||||
|
||||
@@ -33,10 +33,11 @@
|
||||
use std::{
|
||||
io,
|
||||
pin::Pin,
|
||||
sync::Arc,
|
||||
task::{Context, Poll},
|
||||
};
|
||||
|
||||
use axum::extract::ws::{CloseFrame, Message, WebSocket};
|
||||
use axum::extract::ws::{CloseFrame, Message as AxumMessage, WebSocket};
|
||||
use futures::channel::mpsc as futures_mpsc;
|
||||
use futures::{SinkExt, StreamExt};
|
||||
use tokio::io::{AsyncRead, AsyncWrite};
|
||||
@@ -80,6 +81,7 @@ pub struct WsByteStream {
|
||||
pub struct WsPumps {
|
||||
read_task: tokio::task::JoinHandle<()>,
|
||||
write_task: tokio::task::JoinHandle<()>,
|
||||
read_eof: Arc<tokio::sync::Notify>,
|
||||
}
|
||||
|
||||
impl WsPumps {
|
||||
@@ -87,6 +89,13 @@ impl WsPumps {
|
||||
self.read_task.abort();
|
||||
self.write_task.abort();
|
||||
}
|
||||
|
||||
/// Fires when the WS read side reaches EOF (socket close from either
|
||||
/// side) — used by `from_wss`'s connection-drop monitor to await
|
||||
/// socket EOF (ADR-070).
|
||||
pub(crate) async fn read_eof(&self) {
|
||||
self.read_eof.notified().await;
|
||||
}
|
||||
}
|
||||
|
||||
/// Split a `WebSocket` into the byte stream + the pump tasks. The
|
||||
@@ -97,26 +106,30 @@ pub fn split_ws_to_bytes(socket: WebSocket) -> (WsByteStream, WsPumps) {
|
||||
let (read_tx, read_rx) = mpsc::channel::<Vec<u8>>(READ_SLOTS);
|
||||
let (write_tx, mut write_rx) = futures_mpsc::channel::<WriteMsg>(WRITE_SLOTS);
|
||||
|
||||
let read_eof = Arc::new(tokio::sync::Notify::new());
|
||||
|
||||
let write_tx_for_read = write_tx.clone();
|
||||
let read_eof_for_task = Arc::clone(&read_eof);
|
||||
let read_task = tokio::spawn(async move {
|
||||
while let Some(msg) = ws_stream.next().await {
|
||||
match msg {
|
||||
Ok(Message::Binary(b)) => {
|
||||
Ok(AxumMessage::Binary(b)) => {
|
||||
if read_tx.send(b.to_vec()).await.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
Ok(Message::Text(_)) => {
|
||||
Ok(AxumMessage::Text(_)) => {
|
||||
let _ = write_tx_for_read
|
||||
.clone()
|
||||
.send(WriteMsg::CloseWith(WS_PROTOCOL_ERROR))
|
||||
.await;
|
||||
break;
|
||||
}
|
||||
Ok(Message::Close(_)) | Err(_) => break,
|
||||
Ok(AxumMessage::Close(_)) | Err(_) => break,
|
||||
Ok(_) => {}
|
||||
}
|
||||
}
|
||||
read_eof_for_task.notify_waiters();
|
||||
});
|
||||
|
||||
let write_task = tokio::spawn(async move {
|
||||
@@ -125,7 +138,7 @@ pub fn split_ws_to_bytes(socket: WebSocket) -> (WsByteStream, WsPumps) {
|
||||
match msg {
|
||||
WriteMsg::CloseWith(code) => {
|
||||
let _ = ws_sink
|
||||
.send(Message::Close(Some(CloseFrame {
|
||||
.send(AxumMessage::Close(Some(CloseFrame {
|
||||
code,
|
||||
reason: "text messages not supported".into(),
|
||||
})))
|
||||
@@ -147,7 +160,7 @@ pub fn split_ws_to_bytes(socket: WebSocket) -> (WsByteStream, WsPumps) {
|
||||
let chunk: Vec<u8> = pending.drain(..total).collect();
|
||||
for piece in chunk.chunks(WS_MESSAGE_CAP) {
|
||||
if ws_sink
|
||||
.send(Message::Binary(piece.to_vec().into()))
|
||||
.send(AxumMessage::Binary(piece.to_vec().into()))
|
||||
.await
|
||||
.is_err()
|
||||
{
|
||||
@@ -171,6 +184,7 @@ pub fn split_ws_to_bytes(socket: WebSocket) -> (WsByteStream, WsPumps) {
|
||||
WsPumps {
|
||||
read_task,
|
||||
write_task,
|
||||
read_eof,
|
||||
},
|
||||
)
|
||||
}
|
||||
@@ -253,3 +267,107 @@ impl AsyncWrite for WsByteStream {
|
||||
Poll::Ready(Ok(()))
|
||||
}
|
||||
}
|
||||
|
||||
/// Client-side split for a tokio-tungstenite `WebSocketStream`
|
||||
/// (`from_wss`, ADR-070): the same WS↔byte-stream seam as
|
||||
/// [`split_ws_to_bytes`], over the tungstenite socket instead of axum's
|
||||
/// server-side `WebSocket`. Message semantics are identical: binary
|
||||
/// messages carry the byte stream, text is a protocol error (close
|
||||
/// 1002), close → read EOF.
|
||||
#[cfg(any(test, feature = "wss"))]
|
||||
pub fn split_tungstenite_to_bytes<S>(
|
||||
socket: tokio_tungstenite::WebSocketStream<S>,
|
||||
) -> (WsByteStream, WsPumps)
|
||||
where
|
||||
S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static,
|
||||
{
|
||||
let (mut ws_sink, mut ws_stream) = socket.split();
|
||||
let (read_tx, read_rx) = mpsc::channel::<Vec<u8>>(READ_SLOTS);
|
||||
let (write_tx, mut write_rx) = futures_mpsc::channel::<WriteMsg>(WRITE_SLOTS);
|
||||
|
||||
let read_eof = Arc::new(tokio::sync::Notify::new());
|
||||
|
||||
let write_tx_for_read = write_tx.clone();
|
||||
let read_eof_for_task = Arc::clone(&read_eof);
|
||||
let read_task = tokio::spawn(async move {
|
||||
while let Some(msg) = ws_stream.next().await {
|
||||
match msg {
|
||||
Ok(tokio_tungstenite::tungstenite::Message::Binary(b)) => {
|
||||
if read_tx.send(b.to_vec()).await.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
Ok(tokio_tungstenite::tungstenite::Message::Text(_)) => {
|
||||
let _ = write_tx_for_read
|
||||
.clone()
|
||||
.send(WriteMsg::CloseWith(WS_PROTOCOL_ERROR))
|
||||
.await;
|
||||
break;
|
||||
}
|
||||
Ok(tokio_tungstenite::tungstenite::Message::Close(_)) | Err(_) => break,
|
||||
Ok(_) => {}
|
||||
}
|
||||
}
|
||||
read_eof_for_task.notify_waiters();
|
||||
});
|
||||
|
||||
let write_task = tokio::spawn(async move {
|
||||
let mut pending: Vec<u8> = Vec::new();
|
||||
while let Some(msg) = write_rx.next().await {
|
||||
match msg {
|
||||
WriteMsg::CloseWith(code) => {
|
||||
let _ = ws_sink
|
||||
.send(tokio_tungstenite::tungstenite::Message::Close(Some(
|
||||
tokio_tungstenite::tungstenite::protocol::CloseFrame {
|
||||
code: code.into(),
|
||||
reason: "text messages not supported".into(),
|
||||
},
|
||||
)))
|
||||
.await;
|
||||
break;
|
||||
}
|
||||
WriteMsg::Bytes(b) => pending.extend_from_slice(&b),
|
||||
}
|
||||
loop {
|
||||
if pending.len() < 8 {
|
||||
break;
|
||||
}
|
||||
let len =
|
||||
u32::from_be_bytes([pending[4], pending[5], pending[6], pending[7]]) as usize;
|
||||
let total = 8 + len;
|
||||
if pending.len() < total {
|
||||
break;
|
||||
}
|
||||
let chunk: Vec<u8> = pending.drain(..total).collect();
|
||||
for piece in chunk.chunks(WS_MESSAGE_CAP) {
|
||||
if ws_sink
|
||||
.send(tokio_tungstenite::tungstenite::Message::Binary(
|
||||
piece.to_vec().into(),
|
||||
))
|
||||
.await
|
||||
.is_err()
|
||||
{
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
let _ = ws_sink.close().await;
|
||||
});
|
||||
|
||||
(
|
||||
WsByteStream {
|
||||
read_rx,
|
||||
read_buf: Vec::new(),
|
||||
read_pos: 0,
|
||||
eof: false,
|
||||
write_tx,
|
||||
write_open: true,
|
||||
},
|
||||
WsPumps {
|
||||
read_task,
|
||||
write_task,
|
||||
read_eof,
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
@@ -13,7 +13,8 @@ pub mod byte_adapter;
|
||||
pub mod upgrade;
|
||||
|
||||
pub use byte_adapter::{
|
||||
split_ws_to_bytes, WsByteStream, WsPumps, WS_MESSAGE_CAP, WS_PROTOCOL_ERROR,
|
||||
split_tungstenite_to_bytes, split_ws_to_bytes, WsByteStream, WsPumps, WS_MESSAGE_CAP,
|
||||
WS_PROTOCOL_ERROR,
|
||||
};
|
||||
pub use upgrade::{run_channels_session, ws_bearer_auth, ws_upgrade_handler};
|
||||
|
||||
|
||||
@@ -35,10 +35,15 @@ pub async fn run_channels_session(
|
||||
) {
|
||||
let (byte_stream, _pumps) = split_ws_to_bytes(socket);
|
||||
let conn = Connection::from_bidi(byte_stream, b"alk/channels".to_vec(), None);
|
||||
let _ = conn.set_identity(identity);
|
||||
let _ = conn.set_identity(identity.clone());
|
||||
|
||||
let adapter = ChannelsAdapter::new(install_channel_zero(registry), policy);
|
||||
let auth = AuthContext::anonymous(b"alk/channels");
|
||||
let auth = AuthContext {
|
||||
identity: Some(identity),
|
||||
alpn: b"alk/channels".to_vec(),
|
||||
remote_addr: None,
|
||||
tls_client_fingerprint: None,
|
||||
};
|
||||
if let Err(e) = ProtocolHandler::handle(&adapter, conn, &auth).await {
|
||||
tracing::warn!(error = %e, "channels session ended");
|
||||
}
|
||||
@@ -57,9 +62,18 @@ pub async fn run_channels_session(
|
||||
fn install_channel_zero(
|
||||
registry: Arc<OperationRegistry>,
|
||||
) -> alkcall::channels::adapter::InstallChannelZero {
|
||||
Arc::new(move |_manager, channel0_conn, _auth| {
|
||||
Arc::new(move |_manager, channel0_conn, auth| {
|
||||
let registry = Arc::clone(®istry);
|
||||
tokio::spawn(async move {
|
||||
// The WS identity rides the upgrade request; propagate it to
|
||||
// channel 0's `CallConnection` so the dispatcher's
|
||||
// `resolve_identity` (and thus `AccessControl::check` on
|
||||
// `services/list` and every operation) sees it. Without this
|
||||
// the freshly constructed channel-0 `Connection` carries no
|
||||
// identity and all ACL-restricted ops look unauthenticated.
|
||||
if let Some(identity) = auth.identity.clone() {
|
||||
let _ = channel0_conn.set_identity(identity);
|
||||
}
|
||||
let channel0_bidi = match channel0_conn.accept_bi().await {
|
||||
Ok(s) => s,
|
||||
Err(_) => return,
|
||||
|
||||
Reference in New Issue
Block a user