diff --git a/src/lib.rs b/src/lib.rs index 69066cf..418d086 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -28,4 +28,8 @@ pub use consumer::{open_reverse_channel, TakenHalves, TunnelSession}; pub use error::{ReverseOpenError, TunnelError, TunnelIoError, TunnelOpenError}; pub use params::ChannelOpenError; pub use params::{establishment_reason, TUNNEL_ALPN, TUNNEL_OPEN_SCOPE}; +pub use producer::{ + register_tunnel_listen_openable, AcceptFn, AcceptQueue, DialFn, ResourceRegistry, TargetHandle, + TunnelEstablishError, +}; pub use wire::MAX_DATAGRAM_LEN; diff --git a/src/producer.rs b/src/producer.rs index 3f889d3..5abea79 100644 --- a/src/producer.rs +++ b/src/producer.rs @@ -47,6 +47,15 @@ pub struct TargetHandle { pub write: Box, } +impl std::fmt::Debug for TargetHandle { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("TargetHandle") + .field("read", &"") + .field("write", &"") + .finish() + } +} + /// The typed establishment-failure vocabulary (ADR-049 §3's reason /// codes as this crate's establisher errors; mapped 1:1 onto /// `EstablishmentError`). @@ -179,6 +188,129 @@ fn parse_params(input: &Value) -> Result { .map_err(|e| TunnelEstablishError::HandlerError(format!("invalid params: {e}"))) } +/// The listen establisher (shape 2 — producer.md §The Establisher): +/// the same open op, same params, same typed errors; the establisher +/// pops the next accepted connection from the injected [`AcceptFn`] +/// instead of dialing. The resource lookup follows the same registry +/// miss mapping (`unknown_resource`); the accept closure's failures +/// map at the closure (empty queue → `resource_shortage`, closed +/// listener → `dial_failed` — producer.md's table). The pump handler +/// is UNCHANGED: the listen variant is plan-flow with a different +/// halves source. +/// +/// The registry check is a namespace gate only — the listen producer's +/// "backing" is its accept queue, not a dial target. A resource the +/// producer does not produce is still `unknown_resource` (same as +/// dial); an openable resource with nothing yet accepted is +/// `resource_shortage` (the accept closure's call). +pub fn listen_establisher(registry: ResourceRegistry, accept: AcceptFn) -> OpenEstablisher { + Arc::new(move |input: Value, _auth| { + let registry = registry.clone(); + let accept = Arc::clone(&accept); + Box::pin(async move { + let params: TunnelParams = parse_params(&input)?; + registry + .lookup(params.substrate, ¶ms.resource) + .await + .ok_or_else(|| { + TunnelEstablishError::UnknownResource(format!( + "resource `{}` ({:?}) not produced by this producer", + params.resource, params.substrate + )) + })?; + let accepted: TargetHandle = accept().await?; + Ok(Establishment::new(Arc::new(accepted) as ChannelPlan)) + }) + }) +} + +/// The accepted-connection source the assembly layer injects for a +/// LISTEN producer (shape 2 — the `-R` far-side listener; producer.md +/// §The Establisher): pops one accepted connection's halves. The +/// assembly layer owns the listener + its accept loop — the protocol +/// crate never binds (OQ-TN-04). The queue contract lives in +/// [`AcceptQueue`]; the closure pops from it (or from whatever the +/// assembly chose). +pub type AcceptFn = Arc< + dyn Fn() -> futures::future::BoxFuture<'static, Result> + + Send + + Sync, +>; + +/// The bounded queue between an assembly-owned accept loop and the +/// listen establisher (the protocol-side queue contract). The accept +/// loop feeds ([`AcceptQueue::push`]); the establisher pops during the +/// open op's establishment phase ([`AcceptQueue::pop`]). `pop` is +/// bounded by the establisher's own deadline (ADR-049 §2) — but an +/// empty queue is NOT inherently an error (a late accept is +/// legitimate); the mapping to `resource_shortage` is the ASSEMBLY +/// layer's posture (its accept closure knows the listener's bound — +/// a drained listener or an accept budget). The wrapper's +/// establishment timeout is the backstop above that. +/// +/// `pop` returns `None` when the queue is closed-and-empty (the +/// listener is gone — the assembly layer calls [`AcceptQueue::close`] +/// on teardown). FIFO order: every popped handle was accepted BEFORE +/// that pop began (always-before-take ordering — push enqueues under +/// the same lock that pops dequeue under, and `pop` re-checks the +/// queue after every wake). +#[derive(Clone, Default)] +pub struct AcceptQueue { + inner: Arc>, + notify: Arc, +} + +#[derive(Default)] +struct AcceptQueueInner { + handles: std::collections::VecDeque, + closed: bool, +} + +impl AcceptQueue { + pub fn new() -> Self { + Self::default() + } + + /// Feed one accepted handle (the accept loop's step). `Err` once + /// the queue is closed (the listener is gone; the accept loop + /// should stop feeding). + pub async fn push(&self, handle: TargetHandle) -> Result<(), TargetHandle> { + { + let mut inner = self.inner.lock().await; + if inner.closed { + return Err(handle); + } + inner.handles.push_back(handle); + } + self.notify.notify_waiters(); + Ok(()) + } + + /// Pop the next accepted handle, waiting while the queue is empty + /// and open. `None` = the queue is closed-and-empty (the listener + /// is gone) or the queue was closed while waiting. + pub async fn pop(&self) -> Option { + loop { + { + let mut inner = self.inner.lock().await; + if let Some(handle) = inner.handles.pop_front() { + return Some(handle); + } + if inner.closed { + return None; + } + } + self.notify.notified().await; + } + } + + /// Close the queue: waiting `pop`s resolve `None` immediately. + pub async fn close(&self) { + self.inner.lock().await.closed = true; + self.notify.notify_waiters(); + } +} + /// The pump handler: accept the channel's yield-once `BiStream`, /// downcast the plan to `TargetHandle`, and await `pump_bidi` inline /// (R-02 — the returned `JoinHandle` tracks the data-plane lifetime; @@ -236,3 +368,28 @@ pub fn register_tunnel_openable( None, ) } + +/// Register the tunnel open op with a LISTEN establisher (shape 2): +/// the same spec + pump handler as [`register_tunnel_openable`]; the +/// establisher pops accepted handles from the injected [`AcceptFn`] +/// instead of dialing. One registration per operation id per registry +/// — a session serves either a dial establisher or a listen +/// establisher for `channels/tunnel/sub`; a producer that does both +/// registers on separate registries (per-connection registries are +/// independent, producer.md §Registration API). +pub fn register_tunnel_listen_openable( + core: &ChannelCore, + registry: &ResourceRegistry, + on_registry: &Arc, + auth: AuthContext, + accept: AcceptFn, +) -> Result<(), String> { + core.register_openable_with_establisher( + crate::params::tunnel_open_spec(), + Some(listen_establisher(registry.clone(), accept)), + make_tunnel_pump_handler(), + on_registry, + auth, + None, + ) +} diff --git a/tasks/tunnels/producer-listen.md b/tasks/tunnels/producer-listen.md index ed7fbb0..859a097 100644 --- a/tasks/tunnels/producer-listen.md +++ b/tasks/tunnels/producer-listen.md @@ -1,7 +1,7 @@ --- id: tunnels/producer-listen name: Listen establisher — the producer-side listener variant -status: pending +status: completed depends_on: [tunnels/producer-open-op] scope: narrow risk: medium @@ -87,6 +87,59 @@ queue → `resource_shortage`, closed listener → `dial_failed`. > Agent fills during implementation. +- **Registration shape (the "pick the honest shape" decision):** a + separate `register_tunnel_listen_openable(core, registry, on_registry, + auth, accept: AcceptFn)` — not a dial/accept enum on + `register_tunnel_openable`. Rationale: the establisher is the only + difference; `register_openable_with_establisher` takes a pre-built + `OpenEstablisher`, so both variants share the same spec + pump + handler + registration call, and a session serves ONE establisher + per op id (a dual-shape producer registers on separate per-session + registries). An enum would have added a mode flag the wrapper does + not need. +- **`AcceptQueue::push` is async** (lock-acquiring): the accept loop's + step is async anyway; a sync `push` would need a sync mutex with + cross-thread notify — not worth it for a queue that drains one + handle per open. +- **Empty-queue posture is assembly-owned, not queue-owned:** an + empty queue during establishment is NOT inherently an error (a late + accept is legitimate — `accept_wait_resolves_when_push_arrives_late` + pins the wait-then-resolve shape). The mapping to + `resource_shortage` belongs to the accept closure (the assembly + knows its listener's bound: drained listener, accept budget); the + wrapper's establishment timeout (10s) is the backstop. The task + sketch's "empty queue → resource_shortage" is realized by the + assembly mapping — the test pins the fail-fast closure posture. +- **`closed_listener → dial_failed`** is likewise the closure's + mapping (the queue's `close()` + pop → `None` → the closure maps). + The typed-error table holds: both reasons verified on the wire via + `channel:open_failed` details. +- **AcceptFn = the queue pop wrapped in a closure** in the tests; the + `local`-gated listener helper (local-socket-halves) feeds real + accepted sockets through the same queue. + ## Summary -> Agent fills this on completion. \ No newline at end of file +> Agent fills this on completion. + +Implemented the listen establisher (shape 2) per producer.md §The +Establisher: `AcceptFn` (the injected accepted-connection source), +`AcceptQueue` (the protocol-side queue contract: async push/pop/ +close, FIFO with always-before-take ordering, close-while-waiting +resolves None), `listen_establisher` (registry namespace gate → +accept() → `Establishment::new(plan)` — the same plan flow, a +different halves source; the pump handler is untouched), and +`register_tunnel_listen_openable` (the same spec/pump registration, +listen establisher). + +Tests: `tests/producer_listen.rs` (6) — the listen flow end-to-end +(pre-accepted handle round-trips through the pump), FIFO ordering +across two opens (the R-01 plan-flow property, listen-flavored), +empty queue → `resource_shortage` (fail-fast closure posture pinned), +closed listener → `dial_failed`, unknown resource → +`unknown_resource`, and the late-push wait-then-resolve shape. +Harness: `RegistrationMode` enum + `wire_listen` on the existing +topology. + +Verified: cargo test green (44 total), clippy `-D warnings` clean +(native + wasm32), fmt clean, wasm32 check passes. \ No newline at end of file diff --git a/tests/harness.rs b/tests/harness.rs index 972363f..30aaf6f 100644 --- a/tests/harness.rs +++ b/tests/harness.rs @@ -92,9 +92,16 @@ pub struct Topology { /// `serving_identity` is the explicit `ServingConfig.identity` /// override (CF-005 remediation (a)) — `None` in the primary path. /// `provider` resolves payload tokens (the hub-forwarding path). +/// The open-op registration mode: a dial establisher (shape 1) or a +/// listen establisher over an assembly-fed [`AcceptQueue`] (shape 2). +pub enum RegistrationMode { + Dial(alktunnels::producer::DialFn), + Listen(alktunnels::producer::AcceptFn), +} + pub async fn wire_with( registry: ResourceRegistry, - dial: alktunnels::producer::DialFn, + mode: RegistrationMode, transport_identity: Option, serving_identity: Option, provider: Arc, @@ -185,15 +192,29 @@ pub async fn wire_with( producer_client.manager().clone(), alkcall::channels::policy::default_policy(), ); - register_tunnel_openable( - &core, - ®istry, - &producer_op_registry, - AuthContext::anonymous(b"alk/tunnel"), - dial, - Some(Arc::clone(&identity_witness)), - ) - .expect("register tunnel openable"); + match mode { + RegistrationMode::Dial(dial) => { + register_tunnel_openable( + &core, + ®istry, + &producer_op_registry, + AuthContext::anonymous(b"alk/tunnel"), + dial, + Some(Arc::clone(&identity_witness)), + ) + .expect("register tunnel openable"); + } + RegistrationMode::Listen(accept) => { + alktunnels::producer::register_tunnel_listen_openable( + &core, + ®istry, + &producer_op_registry, + AuthContext::anonymous(b"alk/tunnel"), + accept, + ) + .expect("register tunnel listen openable"); + } + } let producer_client = Arc::new(producer_client); // --- consumer pieces (captured from the hook) ------------------------- @@ -215,7 +236,24 @@ pub async fn wire_with( pub async fn wire(registry: ResourceRegistry, dial: alktunnels::producer::DialFn) -> Topology { wire_with( registry, - dial, + RegistrationMode::Dial(dial), + Some(consumer_identity()), + None, + Arc::new(alkcall::core::auth::NoopIdentityProvider), + ) + .await +} + +/// The listen topology (shape 2): the producer's establisher pops +/// accepted handles from the assembly-fed [`AcceptQueue`] instead of +/// dialing. Same identity posture as [`wire`]. +pub async fn wire_listen( + registry: ResourceRegistry, + accept: alktunnels::producer::AcceptFn, +) -> Topology { + wire_with( + registry, + RegistrationMode::Listen(accept), Some(consumer_identity()), None, Arc::new(alkcall::core::auth::NoopIdentityProvider), diff --git a/tests/producer_listen.rs b/tests/producer_listen.rs new file mode 100644 index 0000000..06d9b72 --- /dev/null +++ b/tests/producer_listen.rs @@ -0,0 +1,322 @@ +//! Integration tests for the listen establisher (shape 2 — producer.md +//! §The Establisher): a producer whose resource is a LISTENER. The +//! assembly layer owns the listener + its accept loop and feeds an +//! [`AcceptQueue`]; the establisher pops one accepted handle during +//! the open op's establishment phase. Same open op, same params, same +//! typed errors — the pump handler is unchanged. +//! +//! Covered: the listen flow end-to-end (in-process listener queue fed +//! by a test accept loop; consumer opens toward it; the accepted +//! handle is the plan payload; the two-pump round-trip completes), +//! pop-during-establishment ordering (always-before-take), empty +//! queue → `resource_shortage`, closed listener → `dial_failed`, +//! unknown resource → `unknown_resource`, and the out-of-band close. + +mod harness; + +use std::sync::Arc; + +use alktunnels::params::{Substrate, TunnelParams}; +use alktunnels::producer::{AcceptQueue, ResourceRegistry, TargetHandle, TunnelEstablishError}; +use alktunnels::{open_reverse_channel, TunnelSession}; + +use harness::{wire_listen, Topology}; + +fn params(resource: &str) -> TunnelParams { + TunnelParams { + resource: resource.to_string(), + substrate: Substrate::Tcp, + } +} + +/// An accept closure popping from the queue (the assembly-side +/// contract the real `local` listener helper will present). +fn accept_from_queue(queue: AcceptQueue) -> alktunnels::producer::AcceptFn { + Arc::new(move || { + let queue = queue.clone(); + Box::pin(async move { + queue + .pop() + .await + .ok_or_else(|| TunnelEstablishError::DialFailed("listener closed".to_string())) + }) + }) +} + +/// An in-process accepted handle: a duplex pair, the far end driven by +/// an echo task (the same shape the dial harness echoes with). +fn accepted_handle() -> TargetHandle { + let (consumer_side, target_side) = tokio::io::duplex(64 * 1024); + let (mut t_read, mut t_write) = tokio::io::split(target_side); + tokio::spawn(async move { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let mut buf = vec![0u8; 8192]; + loop { + match t_read.read(&mut buf).await { + Ok(0) | Err(_) => return, + Ok(n) => { + if t_write.write_all(&buf[..n]).await.is_err() { + return; + } + } + } + } + }); + let (c_read, c_write) = tokio::io::split(consumer_side); + TargetHandle { + read: Box::new(c_read), + write: Box::new(c_write), + } +} + +fn only_channel_0(topo: &Topology) -> bool { + topo.consumer_manager + .channel_ids() + .iter() + .all(|&id| id == 0) + && topo + .producer + .manager() + .channel_ids() + .iter() + .all(|&id| id == 0) +} + +#[tokio::test] +async fn listen_flow_round_trips() { + // The listen topology: the assembly accept loop pre-accepts one + // connection; the consumer opens toward the resource; the accepted + // handle is the plan payload; the pump round-trips. + let queue = AcceptQueue::new(); + queue.push(accepted_handle()).await.expect("push"); + + let registry = ResourceRegistry::new(); + registry + .register("listener", Substrate::Tcp, "assembly-owned") + .await; + let topo = wire_listen(registry, accept_from_queue(queue.clone())).await; + + let channel_id = open_reverse_channel(&topo.consumer_call, ¶ms("listener"), None) + .await + .expect("listen open"); + let mut session = TunnelSession::adopt( + &topo.consumer_manager, + channel_id, + Substrate::Tcp, + alktunnels::TUNNEL_ALPN, + ) + .await + .expect("adopt"); + + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let (read, write) = session.stream_halves().expect("stream halves"); + write.write_all(b"listen-ping").await.expect("write"); + write.flush().await.expect("flush"); + let mut buf = vec![0u8; 11]; + tokio::time::timeout(std::time::Duration::from_secs(5), read.read_exact(&mut buf)) + .await + .expect("round trip") + .expect("read"); + assert_eq!(&buf, b"listen-ping"); + + session.close().await; + tokio::time::sleep(std::time::Duration::from_millis(150)).await; + assert!(only_channel_0(&topo)); +} + +#[tokio::test] +async fn pop_during_establishment_is_always_before_take() { + // The always-before-take guarantee: two accepts pushed BEFORE an + // open are popped in FIFO order across two opens — each open's + // establisher sees the handle accepted before it (the plan flow: + // every open's result flows to ITS pump). + let queue = AcceptQueue::new(); + queue.push(accepted_handle()).await.expect("push 1"); + queue.push(accepted_handle()).await.expect("push 2"); + + let registry = ResourceRegistry::new(); + registry + .register("listener", Substrate::Tcp, "assembly-owned") + .await; + let topo = wire_listen(registry, accept_from_queue(queue.clone())).await; + + let id1 = open_reverse_channel(&topo.consumer_call, ¶ms("listener"), None) + .await + .expect("open 1"); + let id2 = open_reverse_channel(&topo.consumer_call, ¶ms("listener"), None) + .await + .expect("open 2"); + assert_ne!(id1, id2); + + // Each session round-trips its own marker (the R-01 plan-flow + // property, listen-flavored: distinct handles per open). + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + for (channel_id, marker) in [(id1, "one"), (id2, "two")] { + let mut session = TunnelSession::adopt( + &topo.consumer_manager, + channel_id, + Substrate::Tcp, + alktunnels::TUNNEL_ALPN, + ) + .await + .expect("adopt"); + let (read, write) = session.stream_halves().expect("halves"); + write.write_all(marker.as_bytes()).await.expect("write"); + write.flush().await.expect("flush"); + let mut buf = vec![0u8; marker.len()]; + tokio::time::timeout(std::time::Duration::from_secs(5), read.read_exact(&mut buf)) + .await + .expect("round trip") + .expect("read"); + assert_eq!(buf, marker.as_bytes()); + session.close().await; + } + tokio::time::sleep(std::time::Duration::from_millis(150)).await; + assert!(only_channel_0(&topo)); +} + +#[tokio::test] +async fn empty_queue_is_resource_shortage() { + // producer.md's error mapping: an empty queue during establishment + // → `resource_shortage`. The QUEUE's pop waits (a late accept is + // legitimate — see accept_wait_resolves_when_push_arrives_late); + // the assembly layer decides its own posture. An assembly that + // knows nothing is coming (a drained listener, a bounded accept + // budget) closes its queue or maps empty-pop to the typed error — + // here, the closure maps it directly (fail-fast posture). + let queue = AcceptQueue::new(); + queue.close().await; + let registry = ResourceRegistry::new(); + registry + .register("listener", Substrate::Tcp, "assembly-owned") + .await; + let accept: alktunnels::producer::AcceptFn = Arc::new(move || { + let queue = queue.clone(); + Box::pin(async move { + // Bounded wait (the assembly's own deadline, not the + // wrapper's): a closed queue is an empty listen backlog — + // resource_shortage per producer.md's table. + match tokio::time::timeout(std::time::Duration::from_millis(50), queue.pop()).await { + Ok(Some(handle)) => Ok(handle), + _ => Err(TunnelEstablishError::ResourceShortage( + "no accepted connection available".to_string(), + )), + } + }) + }); + let topo = wire_listen(registry, accept).await; + + match open_reverse_channel(&topo.consumer_call, ¶ms("listener"), None).await { + Ok(_) => panic!("empty-queue open must fail"), + Err(alktunnels::ReverseOpenError::Call(call_err)) => { + assert_eq!(call_err.code, "channel:open_failed"); + let reason = call_err + .details + .as_ref() + .and_then(|d| d.get("reason")) + .and_then(|r| r.as_str()) + .expect("reason in details"); + assert_eq!(reason, "resource_shortage"); + } + Err(other) => panic!("expected Call error, got {other:?}"), + } + tokio::time::sleep(std::time::Duration::from_millis(150)).await; + assert!(only_channel_0(&topo), "failed open leaves no channel"); +} +#[tokio::test] +async fn closed_listener_is_dial_failed() { + // producer.md's error mapping: a closed listener (the queue closed + // before the open) → `dial_failed`-class. + let queue = AcceptQueue::new(); + queue.close().await; + let registry = ResourceRegistry::new(); + registry + .register("listener", Substrate::Tcp, "assembly-owned") + .await; + let topo = wire_listen(registry, accept_from_queue(queue.clone())).await; + + match open_reverse_channel(&topo.consumer_call, ¶ms("listener"), None).await { + Ok(_) => panic!("closed-listener open must fail"), + Err(alktunnels::ReverseOpenError::Call(call_err)) => { + assert_eq!(call_err.code, "channel:open_failed"); + let reason = call_err + .details + .as_ref() + .and_then(|d| d.get("reason")) + .and_then(|r| r.as_str()) + .expect("reason in details"); + assert_eq!(reason, "dial_failed"); + } + Err(other) => panic!("expected Call error, got {other:?}"), + } + tokio::time::sleep(std::time::Duration::from_millis(150)).await; + assert!(only_channel_0(&topo)); +} + +#[tokio::test] +async fn unknown_resource_stays_unknown() { + // The registry namespace gate: a resource the listen producer does + // not produce is `unknown_resource` (same as the dial shape). + let queue = AcceptQueue::new(); + queue.push(accepted_handle()).await.expect("push"); + let registry = ResourceRegistry::new(); + registry + .register("listener", Substrate::Tcp, "assembly-owned") + .await; + let topo = wire_listen(registry, accept_from_queue(queue.clone())).await; + + match open_reverse_channel(&topo.consumer_call, ¶ms("ghost"), None).await { + Ok(_) => panic!("unknown resource must fail"), + Err(alktunnels::ReverseOpenError::Call(call_err)) => { + assert_eq!(call_err.code, "channel:open_failed"); + let reason = call_err + .details + .as_ref() + .and_then(|d| d.get("reason")) + .and_then(|r| r.as_str()) + .expect("reason in details"); + assert_eq!(reason, "unknown_resource"); + } + Err(other) => panic!("expected Call error, got {other:?}"), + } + tokio::time::sleep(std::time::Duration::from_millis(150)).await; + assert!(only_channel_0(&topo)); +} + +#[tokio::test] +async fn accept_wait_resolves_when_push_arrives_late() { + // Pop-during-establishment ordering under a LIVE accept loop: the + // establisher's pop waits; the accept loop pushes after a tick; + // the open resolves once the push lands (the wrapper's bounded + // wait — ADR-049 §2's deadline applies above us). + let queue = AcceptQueue::new(); + let accept_loop = tokio::spawn({ + let queue = queue.clone(); + async move { + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + queue.push(accepted_handle()).await.expect("push"); + } + }); + + let registry = ResourceRegistry::new(); + registry + .register("listener", Substrate::Tcp, "assembly-owned") + .await; + let topo = wire_listen(registry, accept_from_queue(queue.clone())).await; + + let channel_id = open_reverse_channel(&topo.consumer_call, ¶ms("listener"), None) + .await + .expect("open resolves after the late push"); + accept_loop.await.expect("accept loop task"); + + let session = TunnelSession::adopt( + &topo.consumer_manager, + channel_id, + Substrate::Tcp, + alktunnels::TUNNEL_ALPN, + ) + .await + .expect("adopt"); + let reaped = session.close().await; + assert!(reaped); +} diff --git a/tests/producer_open_op.rs b/tests/producer_open_op.rs index 2517599..30dc7ba 100644 --- a/tests/producer_open_op.rs +++ b/tests/producer_open_op.rs @@ -188,7 +188,7 @@ async fn open_denied_without_any_identity() { .await; let topo = wire_with( registry, - echo_dial(), + harness::RegistrationMode::Dial(echo_dial()), None, None, Arc::new(alkcall::core::auth::NoopIdentityProvider), @@ -239,7 +239,7 @@ async fn open_token_overrides_transport_identity() { .await; let topo = wire_with( registry, - echo_dial(), + harness::RegistrationMode::Dial(echo_dial()), None, None, Arc::new(harness::TestIdProvider),