Repository: capnproto/capnproto-rust (language: rust)

Issue:
Support for "unwrapping" clients to get at the underlying server.
I'm experimenting with implementing Sandstorm's `WebSite` interface from `web-publishing.capnp`, and am unsure how to deal with the way uploading large blobs works. See:

https://github.com/sandstorm-io/sandstorm/blob/d366557c18faa085410334908a240ec88f97910b/src/sandstorm/web-publishing.capnp#L45-L46

and:

https://github.com/sandstorm-io/sandstorm/blob/d366557c18faa085410334908a240ec88f97910b/src/sandstorm/web-publishing.capnp#L81-L88

The interface suggests a flow where:

1. `uploadBlob` hands back a bytestream and blob to the caller
2. The caller writes the data to the bytestream
3. The caller passes the blob back to the server
4. The server then somehow identifies that this is one of the blobs it had previously handed out.

The last point suggests needing to be able to 'unwrap' a client to get at an underlying server (if the server is inside the local vat). I don't know how to do this this with the rust bindings, and am not sure whether it is in fact possible? Afaik the Go implementation doesn't do this either, and I just added this to the Haskell implementation a couple days ago. I'm not actually 100% sure whether the C++ implementation can do this, or if so, how (@kentonv?)

Can the rust implementation do this? If not, how difficult do you think it would be to add?


Discussion / hints:
In C++ you can do this using `CapabilityServerSet`.
There is no equivalent of `CapabilityServerSet` yet in Rust. I don't know how difficult it would be to add.
[`CapabilityServerSet`](https://github.com/capnproto/capnproto/blob/1db1e9f01a8d9dc065b58b8c47b16fbded86859c/c%2B%2B/src/capnp/capability.h#L527-L553) requires downcasting, which Rust does not support in the way that C++ does.

In the case of `web-publishing.capnp`, I think we can get away with something simpler. Instead of adding a way to recover the `Server` object, we could add a way to recover some associated data, which could either be a key identifying the capability, or a shared reference to some of the capability's inner state.

```rust
struct CapabilityAssociatedDataSet<T> {
   // ...
}

impl <T> CapabilityAssociatedDataSet<T> {
    /// Creates a new capability from a `Server` and some associated data.
    pub fn create<C: FromClientHook>(s: Box<dyn capability::Server>, data: T) -> C;

    /// Retrieves data assocated with a capability that was created with this
    /// CapabilityAssociatedDataSet. If the capability was not created with this
    /// set, returns `None.`
    pub async fn get_data(c: &capability::Client) -> Option<&T>;
}
```

There are some details to work out, but I think this would be a fairly straightforward feature to add.

Rust doesn't support downcasting? What are you supposed to do instead?

> we could add a way to recover some associated data, which could either be a key identifying the capability, or a shared reference to some of the capability's inner state.

Why can't be associated data be a reference to the server object itself?

Indeed, I suspect it's at least possible to make the API as complete as
it is in C++ with no rtti -- per the docs downcasting is not possible
there either, but it's still possible to unwrap a client as the type
parameter for the CapabilityServerSet.

Quoting Kenton Varda (2020-04-28 19:39:16)
>    Rust doesn't support downcasting? What are you supposed to do instead?
>
>      we could add a way to recover some associated data, which could
>      either be a key identifying the capability, or a shared reference to
>      some of the capability's inner state.
>
>    Why can't be associated data be a reference to the server object
>    itself?
>
>    --
>    You are receiving this because you authored the thread.
>    Reply to this email directly, [1]view it on GitHub, or [2]unsubscribe.
>
> Verweise
>
>    1. https://github.com/capnproto/capnproto-rust/issues/172#issuecomment-620909863
>    2. https://github.com/notifications/unsubscribe-auth/AAGXYPT7C5HXVBOU7STITJDRO5SKJANCNFSM4MSLL7WA

To be fair, the C++ implementation doesn't need RTTI because the implementation already knows through other means whether the `Client` came from a particular `CapabilityServerSet`, so it can perform a static (rather than dynamic) downcast. But I would expect that a static downcast is "unsafe" in Rust.
> Why can't be associated data be a reference to the server object itself?

You can get a reference to the `dyn capability::Server`, but you can't convert that into a reference to the underlying concrete type.
Can we make use of `Any` for this?

https://doc.rust-lang.org/std/any/trait.Any.html

The Haskell implementation does something similar with Haskell's `Typeable`; perhaps local clients could hang on to a `dyn Any` in some fashion? We may not even need the set part of this.
OK, I think I see a way to make `CapabilityServerSet` work. I put up some first steps in #174.

The new `IntoClient` trait should make it possible for a `CapabilityServerSet` to get a handle on a `ServerDispatch` object, which it can then wrap in an `Rc<Refcell<>>`, passing one ref on to make a capability client, and retaining one ref to allow `get_local_server()` to be implemented.
A crucial finesse here is that the `Rc<Refcell<>>` should wrap the thing (i.e. `ServerDispatch`) implementing the base `capnp::capability::Server` trait, rather than the thing implementing the particular `foo::Server` trait from the generated code. Then we can use a blanket impl:

```rust 
impl <S: capnp::capability::Server> capnp::capability::Server for Rc<RefCell<S>> { ... }
```
whereas doing something similar with the `foo::Server` trait would require generating a bunch more code to forward each method.
I've landed #174. The new trait is now `FromServer`, not `IntoClient` (type inference works out better that way):
https://github.com/capnproto/capnproto-rust/blob/4271dca53868927cc580cc06df92aa6d6ac637c6/capnp/src/capability.rs#L254-L260
Nice, looks like this will make the API more ergonomic too.
Hm, I tried updating my existing code to use the new API, but it's giving me errors like:

```
error[E0277]: the trait bound `sandstorm::util_capnp::assignable::setter::Client<capnp::struct_list::Owned<sandstorm::web_publishing_capnp::web_site::entity::Owned>>: capnp::capability::FromServer<lmdb_web_site::EntitiesCell>` is not satisfied
   --> src/lmdb_web_site.rs:91:34
    |
91  |         results.get().set_setter(capnp_rpc::new_client(ret));
    |                                  ^^^^^^^^^^^^^^^^^^^^^^^^^^ the trait `capnp::capability::FromServer<lmdb_web_site::EntitiesCell>` is not implemented for `sandstorm::util_capnp::assignable::setter::Client<capnp::struct_list::Owned<sandstorm::web_publishing_capnp::web_site::entity::Owned>>`
    | 
   ::: /home/isd/.cargo/registry/src/github.com-1ecc6299db9ec823/capnp-rpc-0.12.2/src/lib.rs:320:45
    |
320 | pub fn new_client<C, S>(s: S) -> C where C: capnp::capability::FromServer<S> {
    |                                             -------------------------------- required by this bound in `capnp_rpc::new_client`
```

The change triggering the error is here:

https://github.com/zenhack/sandstorm-webpub-sysgrain/commit/672139d208fe3c87593fe70a1c7707c14e990bc1

Any idea what's going on? I'm at a loss.
Try `cargo update -p capnpc`.
That did the trick, thanks.

Fix patch:
diff --git a/capnp-rpc/src/lib.rs b/capnp-rpc/src/lib.rs
index e19b570b7..66a9040e1 100644
--- a/capnp-rpc/src/lib.rs
+++ b/capnp-rpc/src/lib.rs
@@ -301,7 +301,63 @@ impl <VatId> Future for RpcSystem<VatId> where VatId: 'static {
 /// Creates a new local RPC client of type `C` out of an object that implements a server trait `S`.
 pub fn new_client<C, S>(s: S) -> C where C: capnp::capability::FromServer<S> {
     capnp::capability::FromClientHook::new(Box::new(
-        local::Client::new(Box::new(<C as capnp::capability::FromServer::<S>>::from_server(s)))))
+        local::Client::new(<C as capnp::capability::FromServer::<S>>::from_server(s))))
+}
+
+/// Allows a server to recognize its own capabilities when passed back to it, and obtain the
+/// underlying Server objects associated with them.
+pub struct CapabilityServerSet<S,C>
+    where C: capnp::capability::FromServer<S>
+{
+    caps: std::collections::HashMap<usize, Rc<RefCell<C::Dispatch>>>,
+}
+
+impl <S,C> CapabilityServerSet<S,C>
+    where C: capnp::capability::FromServer<S>
+{
+    pub fn new() -> Self {
+        Self { caps: std::default::Default::default() }
+    }
+
+    /// Adds a new capability to the set and returns a client backed by it.
+    pub fn new_client(&mut self, s: S) -> C {
+        let dispatch =
+            <C as capnp::capability::FromServer::<S>>::from_server(s);
+        let wrapped = Rc::new(RefCell::new(dispatch));
+        let ptr = wrapped.as_ptr() as usize;
+        self.caps.insert(ptr, wrapped.clone());
+        capnp::capability::FromClientHook::new(Box::new(
+            local::Client::from_rc(wrapped)))
+    }
+
+    /// Looks up a capability and returns its underlying server object, if found.
+    /// Fully resolves the capability before looking it up.
+    pub async fn get_local_server(&self,
+                                  client: &C)
+                                  -> Option<&Rc<RefCell<C::Dispatch>>>
+        where C: capnp::capability::FromClientHook
+    {
+        let resolved: C = capnp::capability::get_resolved_cap(
+            capnp::capability::FromClientHook::new(client.as_client_hook().add_ref())).await;
+        let hook = resolved.into_client_hook();
+        let ptr = hook.get_ptr();
+        self.caps.get(&ptr)
+    }
+
+    /// Looks up a capability and returns its underlying server object, if found.
+    /// Does *not* attempt to resolve the capability first, so you will usually want
+    /// to call `get_resolved_cap()` before calling this. The advantage of this method
+    /// over `get_local_server()` is that this one is synchronous and borrows `self`
+    /// over a shorter span (which can be very important if `self` is inside a `RefCell`).
+    pub fn get_local_server_of_resolved(&self,
+                                        client: &C)
+                                  -> Option<&Rc<RefCell<C::Dispatch>>>
+        where C: capnp::capability::FromClientHook
+    {
+        let hook = client.as_client_hook();
+        let ptr = hook.get_ptr();
+        self.caps.get(&ptr)
+    }
 }
 
 /// Converts a promise for a client into a client that queues up any calls that arrive
diff --git a/capnp-rpc/src/local.rs b/capnp-rpc/src/local.rs
index 7d097b5ca..53a819e32 100644
--- a/capnp-rpc/src/local.rs
+++ b/capnp-rpc/src/local.rs
@@ -278,29 +278,29 @@ impl PipelineHook for Pipeline {
     }
 }
 
-struct ClientInner {
-    server: Box<dyn capability::Server>,
+pub struct Client<S> where S: capability::Server {
+    inner: Rc<RefCell<S>>,
 }
 
-pub struct Client {
-    inner: Rc<RefCell<ClientInner>>,
-}
-
-impl Client {
-    pub fn new(server: Box<dyn capability::Server>) -> Client {
+impl <S> Client<S> where S: capability::Server {
+    pub fn new(server: S) -> Client<S> {
         Client {
-            inner: Rc::new(RefCell::new(ClientInner { server: server }))
+            inner: Rc::new(RefCell::new(server))
         }
     }
+
+    pub fn from_rc(inner: Rc<RefCell<S>>) -> Client<S> {
+        Client { inner }
+    }
 }
 
-impl Clone for Client {
-    fn clone(&self) -> Client {
+impl <S> Clone for Client<S> where S: capability::Server {
+    fn clone(&self) -> Client<S> {
         Client { inner: self.inner.clone() }
     }
 }
 
-impl ClientHook for Client {
+impl <S> ClientHook for Client<S> where S: capability::Server + 'static {
     fn add_ref(&self) -> Box<dyn ClientHook> {
         Box::new(self.clone())
     }
@@ -326,7 +326,7 @@ impl ClientHook for Client {
             let f = {
                 // We put this borrow_mut() inside a block to avoid a potential
                 // double borrow during f.await
-                let server = &mut inner.borrow_mut().server;
+                let server = &mut *inner.borrow_mut();
                 server.dispatch_call(interface_id, method_id,
                                      ::capnp::capability::Params::new(params),
                                      ::capnp::capability::Results::new(results))
@@ -336,7 +336,7 @@ impl ClientHook for Client {
     }
 
     fn get_ptr(&self) -> usize {
-        (&*self.inner.borrow()) as * const _ as usize
+        self.inner.as_ptr() as usize
     }
 
     fn get_brand(&self) -> usize {
diff --git a/capnp/src/capability.rs b/capnp/src/capability.rs
index 780994d13..8a1300d18 100644
--- a/capnp/src/capability.rs
+++ b/capnp/src/capability.rs
@@ -279,3 +279,14 @@ pub trait FromServer<S> : FromClientHook {
 
     fn from_server(s: S) -> Self::Dispatch;
 }
+
+/// Gets the "resolved" version of a capability. One place this is useful is for pre-resolving
+/// the argument to `capnp_rpc::CapabilityServerSet::get_local_server_of_resolved()`.
+pub async fn get_resolved_cap<C: FromClientHook>(cap: C) -> C {
+    let mut hook = cap.into_client_hook();
+    let _ = hook.when_resolved().await;
+    while let Some(resolved) = hook.get_resolved() {
+        hook = resolved;
+    }
+    FromClientHook::new(hook)
+}
diff --git a/capnp/src/private/capability.rs b/capnp/src/private/capability.rs
index 01c2d3d0d..86f22cd4c 100644
--- a/capnp/src/private/capability.rs
+++ b/capnp/src/private/capability.rs
@@ -50,7 +50,11 @@ pub trait ClientHook {
             params: Box<dyn ParamsHook>, results: Box<dyn ResultsHook>)
             -> crate::capability::Promise<(), crate::Error>;
 
+    /// If this capability is associated with an rpc connection, then this method
+    /// returns an identifier for that connection.
     fn get_brand(&self) -> usize;
+
+    /// Returns a (locally) unique identifier for this capabilitiy.
     fn get_ptr(&self) -> usize;
 
     /// If this ClientHook is a promise that has already resolved, returns the inner, resolved version


Test patch:
diff --git a/capnp-rpc/test/impls.rs b/capnp-rpc/test/impls.rs
index eb3de2e47..2572a83c9 100644
--- a/capnp-rpc/test/impls.rs
+++ b/capnp-rpc/test/impls.rs
@@ -20,7 +20,7 @@
 // THE SOFTWARE.
 
 use crate::test_capnp::{bootstrap, test_handle, test_interface, test_extends, test_pipeline,
-                        test_call_order, test_more_stuff};
+                        test_call_order, test_more_stuff, test_capability_server_set};
 
 
 use capnp::Error;
@@ -29,7 +29,7 @@ use capnp_rpc::pry;
 
 use futures::{FutureExt, TryFutureExt};
 
-use std::cell::Cell;
+use std::cell::{Cell, RefCell};
 use std::rc::Rc;
 
 pub struct Bootstrap;
@@ -96,6 +96,14 @@ impl bootstrap::Server for Bootstrap {
         }
         Promise::ok(())
     }
+    fn test_capability_server_set(&mut self,
+                                  _params: bootstrap::TestCapabilityServerSetParams,
+                                  mut results: bootstrap::TestCapabilityServerSetResults)
+                                  -> Promise<(), Error>
+    {
+        results.get().set_cap(capnp_rpc::new_client(TestCapabilityServerSet::new()));
+        Promise::ok(())
+    }
 }
 
 pub struct TestInterface {
@@ -584,3 +592,56 @@ impl test_interface::Server for TestCapDestructor {
     }
 }
 
+pub struct CssHandle {}
+
+impl CssHandle {
+    pub fn new() -> Self { Self {} }
+}
+
+impl test_capability_server_set::handle::Server for CssHandle {}
+
+pub struct TestCapabilityServerSet {
+    set: Rc<RefCell<
+            capnp_rpc::CapabilityServerSet<CssHandle,
+                                           test_capability_server_set::handle::Client>>>,
+}
+
+impl TestCapabilityServerSet {
+    pub fn new() -> Self {
+        Self {
+            set: Rc::new(RefCell::new(capnp_rpc::CapabilityServerSet::new()))
+        }
+    }
+}
+
+impl test_capability_server_set::Server for TestCapabilityServerSet {
+    fn create_handle(&mut self,
+                     _: test_capability_server_set::CreateHandleParams,
+                     mut results: test_capability_server_set::CreateHandleResults)
+                     -> Promise<(), Error>
+    {
+        results.get().set_handle(self.set.borrow_mut().new_client(CssHandle::new()));
+        Promise::ok(())
+    }
+
+    fn check_handle(&mut self,
+                     params: test_capability_server_set::CheckHandleParams,
+                     mut results: test_capability_server_set::CheckHandleResults)
+                     -> Promise<(), Error>
+    {
+        let set = self.set.clone();
+        let handle = pry!(pry!(params.get()).get_handle());
+        Promise::from_future(async move {
+            let resolved = capnp::capability::get_resolved_cap(handle).await;
+            match set.borrow().get_local_server_of_resolved(&resolved) {
+                None => (),
+                Some(_) => {
+                    results.get().set_is_ours(true)
+                }
+            }
+            Ok(())
+        })
+    }
+}
+
+
diff --git a/capnp-rpc/test/test.capnp b/capnp-rpc/test/test.capnp
index 7f61d8ca6..26efb9bd6 100644
--- a/capnp-rpc/test/test.capnp
+++ b/capnp-rpc/test/test.capnp
@@ -78,6 +78,7 @@ interface Bootstrap {
   testPipeline @3 () -> (cap: TestPipeline);
   testCallOrder @4 () -> (cap: TestCallOrder);
   testMoreStuff @5 () -> (cap: TestMoreStuff);
+  testCapabilityServerSet @6 () -> (cap: TestCapabilityServerSet);
 }
 
 interface TestInterface {
@@ -171,3 +172,10 @@ interface TestMoreStuff extends(TestCallOrder) {
   callEachCapability @13 (caps :List(TestInterface)) -> ();
   # Calls TestInterface::foo(123, true) on each cap.
 }
+
+interface TestCapabilityServerSet {
+  interface Handle {}
+
+  createHandle @0 () -> (handle :Handle);
+  checkHandle @1 (handle: Handle) -> (isOurs :Bool);
+}
diff --git a/capnp-rpc/test/test.rs b/capnp-rpc/test/test.rs
index f229329e5..6fa1f08df 100644
--- a/capnp-rpc/test/test.rs
+++ b/capnp-rpc/test/test.rs
@@ -830,3 +830,86 @@ fn capability_list() {
     })
 }
 
+#[test]
+fn capability_server_set() {
+    use capnp_rpc::CapabilityServerSet;
+    use crate::test_capnp::test_interface;
+    use crate::impls;
+    let mut set1 : CapabilityServerSet<impls::TestInterface, test_interface::Client> =
+        CapabilityServerSet::new();
+    let mut set2 : CapabilityServerSet<impls::TestInterface, test_interface::Client> =
+        CapabilityServerSet::new();
+
+    let client_standalone = capnp_rpc::new_client(impls::TestInterface::new());
+
+    let own_server1 = impls::TestInterface::new();
+    let own_server1_counter = own_server1.get_call_count();
+    let client1 = set1.new_client(own_server1);
+
+    let own_server2 = impls::TestInterface::new();
+    let own_server2_counter = own_server2.get_call_count();
+    let client2 = set2.new_client(own_server2);
+
+    // Getting the local server using the correct set works.
+    let own_server1_again = futures::executor::block_on(set1.get_local_server(&client1)).unwrap();
+    assert_eq!(own_server1_again.borrow().get_call_count().as_ptr(), own_server1_counter.as_ptr());
+
+    let own_server2_again = futures::executor::block_on(set2.get_local_server(&client2)).unwrap();
+    assert_eq!(own_server2_again.borrow().get_call_count().as_ptr(), own_server2_counter.as_ptr());
+
+    // Getting the local server using the wrong set doesn't work.
+    assert!(futures::executor::block_on(set1.get_local_server(&client2)).is_none());
+    assert!(futures::executor::block_on(set2.get_local_server(&client1)).is_none());
+    assert!(futures::executor::block_on(set1.get_local_server(&client_standalone)).is_none());
+    assert!(futures::executor::block_on(set2.get_local_server(&client_standalone)).is_none());
+
+    // Also works if the client is a promise.
+    let (fulfiller, promise) = oneshot::channel();
+    let client_promise: test_interface::Client =
+        ::capnp_rpc::new_promise_client(promise.map_err(canceled_to_error));
+
+    let client_promise2: test_interface::Client = client_promise.clone();
+
+    let (error_fulfiller, error_promise) = oneshot::channel();
+    let error_promise: test_interface::Client =
+        ::capnp_rpc::new_promise_client(error_promise.map_err(canceled_to_error));
+
+    assert!(fulfiller.send(client1.client).is_ok());
+    let own_server1_again2 =
+        futures::executor::block_on(set1.get_local_server(&client_promise)).unwrap();
+    assert_eq!(own_server1_again2.borrow().get_call_count().as_ptr(), own_server1_counter.as_ptr());
+
+    // Wrong set; returns None.
+    assert!(futures::executor::block_on(set2.get_local_server(&client_promise2)).is_none());
+
+    drop(error_fulfiller);
+    assert!(futures::executor::block_on(set1.get_local_server(&error_promise)).is_none());
+}
+
+#[test]
+fn capability_server_set_rpc() {
+    rpc_top_level(|_spawner, client| async move {
+        let response1 = client.test_capability_server_set_request().send().promise.await?;
+        let client1 = response1.get()?.get_cap()?;
+
+        let response2 = client.test_capability_server_set_request().send().promise.await?;
+        let client2 = response2.get()?.get_cap()?;
+
+        let handle1 = client1.create_handle_request().send().pipeline.get_handle();
+        let handle2 = client2.create_handle_request().send().pipeline.get_handle();
+
+        // client 1 owns handle 1
+        let mut check_request1 = client1.check_handle_request();
+        check_request1.get().set_handle(handle1.clone());
+        let check_response1 = check_request1.send().promise.await?;
+        assert_eq!(true, check_response1.get()?.get_is_ours());
+
+        // client 1 does not own handle 2
+        let mut check_request2 = client1.check_handle_request();
+        check_request2.get().set_handle(handle2.clone());
+        let check_response2 = check_request2.send().promise.await?;
+        assert_eq!(false, check_response2.get()?.get_is_ours());
+
+        Ok(())
+    })
+}