feat(review 006 Unit 1): channel-open establishment phase + typed client error (E-01, N-1)

Implements ADR-049 Unit 1 — the open-op wrapper gains an awaited,
bounded establishment phase, and the client stops erasing the error.

- OpenEstablisher hook + Establishment/EstablishmentError types:
  register_openable_with_establisher awaits the establisher bounded
  (earlier of dispatch deadline and per-registration timeout, else
  ESTABLISHMENT_TIMEOUT = 10s) after allocation, before the reply and
  before the pump handler is spawned (ADR-049 §1/§2). Implementation
  note: the establisher takes (input, auth) only — the channel's
  yield-once BiStream belongs exclusively to the pump handler
  (amendment recorded in ADR-049).
- Establishment failure: teardown_channel + opener-ledger take +
  policy.on_close un-increment (allocation and teardown balance;
  the ledger take is the atomic gate, ADR-047 §7), reply
  channel:open_failed with details {reason, message} — reason ∈
  dial_failed / unknown_resource / resource_shortage / handler_error
  / timeout (ADR-049 §3). SSH contract consumer-visible: a failed
  open never returns a channel_id.
- register_openable unchanged (no establisher = always-OK; existing
  registrations compile and behave identically — compat gate test).
- ChannelClient::open_channel returns ChannelOpenError (breaking at
  0.5.0): CallFailed { error: CallError } carries the wire error
  verbatim (establishment_reason() branches on details.reason);
  MissingChannelId / AdoptFailed cover the local-only shapes
  (ADR-049 §4, review 006 N-1).
- Tests cover all four verification gates from the review: e2e
  establisher failure through a real channels connection (typed
  reason + no-channel + ledger un-increment), bounded timeout,
  no-establisher compat, establisher-success pump round-trip; plus
  reason-vocabulary mapping and bound arithmetic.
- Bump to 0.5.0 (open_channel error-type change is semver-relevant).

Verification: cargo test (608 passed), clippy --all-targets -D
warnings, fmt --check, doc --no-deps, wasm32 check — all clean.
This commit is contained in:
glm-5.3-flash committed 2026-09-06 19:00:04 +00:00
1 parent 48ceeba55c
commit 2586c3b217
9 files changed
+1106 -31

No files matched your search

+323 -6
View File
@@ -25,13 +25,69 @@ use tokio::sync::Mutex;
use crate::core::types::{Connection, StreamError};
use crate::protocol::connection::CallConnection;
use crate::protocol::wire::ResponseEnvelope;
use crate::protocol::wire::{CallError, ResponseEnvelope};
use crate::registry::registration::OperationRegistry;
use super::manager::{ChannelManager, ChannelSide};
use super::mux::MuxRunner;
use super::reassembly::{MpscRecvStream, MpscSendStream};
/// The typed error from [`ChannelClient::open_channel`] (ADR-049 §4 —
/// review 006 N-1). The open op's `CallError` — including the
/// `channel:open_failed` code with `details: { reason, message }`
/// (ADR-049 §3) — is carried verbatim instead of flattened into a
/// string, so the consumer can branch on the establishment-failure
/// reason (`dial_failed`, `unknown_resource`, `resource_shortage`,
/// `handler_error`, `timeout`).
#[derive(Debug, Clone, PartialEq, thiserror::Error)]
pub enum ChannelOpenError {
/// The open op failed — the `CallError` carries the wire code and
/// typed `details` (`channel:open_failed` + `details.reason` per
/// ADR-049 §3, `channel:too_many_channels`, `FORBIDDEN`, …).
#[error("open op failed: {error:?}")]
CallFailed { error: CallError },
/// The open op succeeded but the response carried no `channel_id`
/// (malformed responder).
#[error("open op response missing channel_id")]
MissingChannelId,
/// The open op succeeded but local channel adoption failed (the
/// per-connection cap, or an ID collision).
#[error("adopt_channel failed: {0}")]
AdoptFailed(#[from] super::manager::ManagerError),
}
impl ChannelOpenError {
/// The wire `CallError` behind this failure, when the open op
/// itself failed. `None` for the local-only variants
/// (`MissingChannelId` comes from a success envelope;
/// `AdoptFailed` is post-reply).
pub fn call_error(&self) -> Option<&CallError> {
match self {
ChannelOpenError::CallFailed { error } => Some(error),
ChannelOpenError::MissingChannelId | ChannelOpenError::AdoptFailed(_) => None,
}
}
/// The establishment-failure reason code — `details.reason` of a
/// `channel:open_failed` error (`dial_failed`, `unknown_resource`,
/// `resource_shortage`, `handler_error`, `timeout`). `None` for
/// any other failure shape.
pub fn establishment_reason(&self) -> Option<&str> {
match self {
ChannelOpenError::CallFailed { error }
if error.code == super::operations::CHANNEL_OPEN_FAILED =>
{
error
.details
.as_ref()
.and_then(|d| d.get("reason"))
.and_then(|r| r.as_str())
}
_ => None,
}
}
}
/// The client-side handle for a channels connection. Constructed via
/// [`ChannelClient::from_connection`] from an established
/// `Connection`. Holds the `ChannelManager` (for the demux/mux state)
@@ -224,26 +280,33 @@ impl ChannelClient {
///
/// `alpn` is the data-plane ALPN (e.g. `alk/tty`), used for
/// observability in the local manager.
///
/// The error is typed (ADR-049 §4 — review 006 N-1): a failed open
/// resolves [`ChannelOpenError::CallFailed`] carrying the
/// wire `CallError` verbatim — branch on
/// [`ChannelOpenError::establishment_reason`] for
/// `channel:open_failed`'s reason code (`dial_failed`,
/// `unknown_resource`, `resource_shortage`, `handler_error`,
/// `timeout`).
pub async fn open_channel(
&self,
operation_id: &str,
input: Value,
alpn: &str,
) -> Result<(u32, MpscSendStream, MpscRecvStream), String> {
) -> Result<(u32, MpscSendStream, MpscRecvStream), ChannelOpenError> {
let response = self.call_open_op(operation_id, input).await;
let out = response
.result
.map_err(|e| format!("open op failed: {e:?}"))?;
.map_err(|error| ChannelOpenError::CallFailed { error })?;
let channel_id = out
.get("channel_id")
.and_then(|v| v.as_u64())
.ok_or_else(|| "open op response missing channel_id".to_string())?
as u32;
.ok_or(ChannelOpenError::MissingChannelId)? as u32;
self.manager
.adopt_channel(channel_id, alpn, None)
.await
.map_err(|e| format!("adopt_channel failed: {e}"))
.map_err(ChannelOpenError::AdoptFailed)
.map(|(send, recv)| (channel_id, send, recv))
}
@@ -1256,6 +1319,260 @@ mod tests {
);
}
/// ADR-049 Unit 1 acceptance gate (review 006 E-01/N-1) — the
/// establisher fails after channel allocation: the consumer's
/// `open_channel` resolves `Err` carrying the wire `CallError`
/// (`channel:open_failed` + `details.reason`), no `channel_id` was
/// ever returned (the SSH "channel never exists opener-side"
/// property, consumer-visible), and the accept side's manager has
/// no channel afterward (teardown + ledger un-increment).
#[tokio::test]
async fn establisher_failure_resolves_typed_open_failed_on_the_client() {
use crate::channels::operations::{
ChannelCore, EstablishmentError, OpenEstablisher, OpenHandler,
};
use crate::channels::policy::PerIdentityChannelPolicy;
use crate::registry::spec::ChannelOpenSpec;
let policy: Arc<PerIdentityChannelPolicy> = Arc::new(PerIdentityChannelPolicy::new(256));
let policy_for_hook: Arc<dyn ChannelLifecyclePolicy> =
Arc::clone(&policy) as Arc<dyn ChannelLifecyclePolicy>;
let policy_for_assert = Arc::clone(&policy);
let open_handler: OpenHandler =
Arc::new(|_input, _channel_conn, _auth| tokio::spawn(async {}));
let establisher: OpenEstablisher = Arc::new(|_input, _auth| {
Box::pin(async {
Err(EstablishmentError::DialFailed {
message: "target refused the connection".to_string(),
})
})
});
let install_hook: crate::channels::adapter::InstallChannelZero =
Arc::new(move |manager, channel0_conn, auth| {
let open_handler = Arc::clone(&open_handler);
let establisher = Arc::clone(&establisher);
let policy = Arc::clone(&policy_for_hook);
tokio::spawn(async move {
let channel0_bidi = match channel0_conn.accept_bi().await {
Ok(s) => s,
Err(_) => return,
};
let (writer, reader) = split_single_stream(channel0_bidi);
let core = ChannelCore::new(manager, policy);
let registry = crate::registry::registration::OperationRegistry::new();
let spec = OperationSpec::new(
"channels/tty/sub",
OperationType::Sub,
Visibility::External,
serde_json::json!({}),
serde_json::json!({
"type": "object",
"properties": { "channel_id": { "type": "integer" } }
}),
vec![],
AccessControl::default(),
None,
)
.with_channel_open(ChannelOpenSpec::new("alk/tty"));
core.register_openable_with_establisher(
spec,
Some(establisher),
open_handler,
&registry,
auth.clone(),
None,
)
.expect("register_openable_with_establisher");
let registry = Arc::new(registry);
let provider: Arc<dyn IdentityProvider> = Arc::new(NoopIdProvider);
let call_connection = Arc::new(CallConnection::new_single_stream(
channel0_conn,
Arc::clone(&writer),
));
let dp = Dispatcher::new(registry, provider);
dp.run_loop_single_stream(call_connection, reader, writer)
.await;
})
});
let (client_end, server_end) = tokio::io::duplex(64 * 1024);
let client_conn =
Connection::from_bidi(client_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
let server_conn =
Connection::from_bidi(server_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
let adapter = ChannelsAdapter::new(install_hook, Arc::new(NoCap));
let auth = AuthContext::anonymous(b"alk/channels");
let _server_handle = tokio::spawn(async move {
let _ = crate::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await;
});
let client = ChannelClient::from_connection(client_conn)
.await
.expect("channel client init");
let err = tokio::time::timeout(
std::time::Duration::from_secs(10),
client.open_channel(
"channels/tty/sub",
serde_json::json!({ "container": "abc" }),
"alk/tty",
),
)
.await
.expect("open_channel timed out");
let err = match err {
Err(e) => e,
Ok(_) => panic!("establisher failure must resolve Err through the client"),
};
// N-1 gate: the typed error carries the wire CallError verbatim.
let call_error = err.call_error().expect("CallFailed carries the CallError");
assert_eq!(call_error.code, "channel:open_failed");
assert_eq!(
err.establishment_reason(),
Some("dial_failed"),
"consumer branches on the typed reason code"
);
// The SSH property, consumer-visible: no channel_id was ever
// returned, and the accept side holds no channel for the failed
// open (channel 0 is the pre-negotiated call channel — always
// present, never the failed open's).
assert!(
client.manager().channel_ids().into_iter().all(|id| id == 0),
"no data channel exists for the failed open"
);
let anonymous = crate::core::auth::Identity {
id: "anonymous".to_string(),
scopes: vec![],
resources: Default::default(),
};
assert_eq!(
policy_for_assert.count_for(&anonymous),
0,
"ledger un-incremented on the accept side after establishment failure"
);
}
/// ADR-049 Unit 1 acceptance gate (companion to the E-01 gate) —
/// establisher success through a real channels connection: the
/// client's `open_channel` adopts and the pumps flow (the happy
/// path is unchanged by the establishment phase).
#[tokio::test]
async fn establisher_success_opens_and_pumps_data_end_to_end() {
use crate::channels::operations::{
ChannelCore, Establishment, OpenEstablisher, OpenHandler,
};
use crate::channels::policy::NoCap;
use crate::registry::spec::ChannelOpenSpec;
use tokio::io::AsyncReadExt;
let (data_tx, mut data_rx) = tokio::sync::mpsc::channel::<Vec<u8>>(1);
let open_handler: OpenHandler = Arc::new(move |_input, channel_conn, _auth| {
let data_tx = data_tx.clone();
tokio::spawn(async move {
let mut bidi = channel_conn.accept_bi().await.expect("accept_bi");
let mut buf = [0u8; 4];
bidi.read_exact(&mut buf).await.expect("read");
data_tx.send(buf.to_vec()).await.expect("send to channel");
})
});
let establisher: OpenEstablisher =
Arc::new(|_input, _auth| Box::pin(async { Ok(Establishment::default()) }));
let establisher_for_hook = Arc::clone(&establisher);
let open_handler_for_hook = Arc::clone(&open_handler);
let install_hook: crate::channels::adapter::InstallChannelZero =
Arc::new(move |manager, channel0_conn, auth| {
let establisher = Arc::clone(&establisher_for_hook);
let open_handler = Arc::clone(&open_handler_for_hook);
tokio::spawn(async move {
let channel0_bidi = match channel0_conn.accept_bi().await {
Ok(s) => s,
Err(_) => return,
};
let (writer, reader) = split_single_stream(channel0_bidi);
let core = ChannelCore::new(manager, Arc::new(NoCap));
let registry = crate::registry::registration::OperationRegistry::new();
let spec = OperationSpec::new(
"channels/tty/sub",
OperationType::Sub,
Visibility::External,
serde_json::json!({
"type": "object",
"properties": { "container": { "type": "string" } },
"required": ["container"]
}),
serde_json::json!({
"type": "object",
"properties": { "channel_id": { "type": "integer" } }
}),
vec![],
AccessControl::default(),
None,
)
.with_channel_open(ChannelOpenSpec::new("alk/tty"));
core.register_openable_with_establisher(
spec,
Some(establisher),
open_handler,
&registry,
auth.clone(),
None,
)
.expect("register_openable_with_establisher");
let registry = Arc::new(registry);
let provider: Arc<dyn IdentityProvider> = Arc::new(NoopIdProvider);
let call_connection = Arc::new(CallConnection::new_single_stream(
channel0_conn,
Arc::clone(&writer),
));
let dp = Dispatcher::new(registry, provider);
dp.run_loop_single_stream(call_connection, reader, writer)
.await;
})
});
let (client_end, server_end) = tokio::io::duplex(64 * 1024);
let client_conn =
Connection::from_bidi(client_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
let server_conn =
Connection::from_bidi(server_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
let adapter = ChannelsAdapter::new(install_hook, Arc::new(NoCap));
let auth = AuthContext::anonymous(b"alk/channels");
let _server_handle = tokio::spawn(async move {
let _ = crate::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await;
});
let client = ChannelClient::from_connection(client_conn)
.await
.expect("channel client init");
let (channel_id, mut send, _recv) = client
.open_channel(
"channels/tty/sub",
serde_json::json!({ "container": "abc" }),
"alk/tty",
)
.await
.expect("open_channel with establisher success");
assert!(channel_id > 0, "channel_id should be non-zero");
send.write_all(b"ping").await.expect("write ping");
drop(send);
let data = tokio::time::timeout(std::time::Duration::from_secs(5), data_rx.recv())
.await
.expect("timed out waiting for handler data")
.expect("handler should receive data");
assert_eq!(&data, b"ping", "handler received ping post-establishment");
}
// --- review 004 Unit 3 acceptance gates (F-04 serving half) -----------
/// F-04 gate 1: hub→consumer call over an existing `ChannelClient`