Implement the Identity Agent

.github/workflows/coverage.yml

@@ -26,7 +26,7 @@ jobs:
       - uses: actions-rs/toolchain@v1
         with:
           profile: minimal
-          toolchain: nightly-2022-01-08
+          toolchain: nightly-2022-04-08
           override: true
           components: llvm-tools-preview
 

.github/workflows/scripts/coverage.sh

@@ -5,7 +5,7 @@ set -e
 rm -rf coverage
 mkdir coverage
 
-NIGHTLY="+nightly-2022-01-08"
+NIGHTLY="+nightly-2022-04-08"
 
 # Run tests with profiling instrumentation
 echo "Running instrumented unit tests..."

Cargo.toml

@@ -5,13 +5,13 @@ members = [
   # "identity-comm",
   "identity-account",
   "identity-account-storage",
+  "identity-actor",
   "identity-core",
   "identity-credential",
   "identity-did",
   "identity-diff",
   "identity-iota",
   "identity-iota-core",
-
   "examples",
 ]
 

identity-actor/Cargo.toml

@@ -0,0 +1,45 @@
+[package]
+name = "identity-actor"
+version = "0.5.0"
+authors = ["IOTA Stiftung"]
+edition = "2021"
+homepage = "https://www.iota.org"
+keywords = ["iota", "tangle", "identity"]
+license = "Apache-2.0"
+readme = "../README.md"
+repository = "https://github.com/iotaledger/identity.rs"
+description = "A peer-to-peer identity communication framework"
+
+[dependencies]
+async-trait = { version = "0.1", default-features = false }
+cfg-if = { version = "1.0", default-features = false }
+dashmap = { version = "5.1", default-features = false }
+futures = { version = "0.3", default-features = false }
+identity-account = { path = "../identity-account", default-features = false, features = ["send-sync-storage"], optional = true }
+identity-core = { path = "../identity-core", default-features = false }
+identity-iota = { path = "../identity-iota", default-features = false }
+identity-iota-core = { path = "../identity-iota-core", default-features = false }
+libp2p = { version = "0.43", default-features = false, features = ["tcp-tokio", "dns-tokio", "websocket", "request-response", "noise", "yamux"] }
+log = { version = "0.4", default-features = false }
+serde = { version = "1.0", default-features = false, features = ["derive"] }
+serde_json = { version = "1.0", default-features = false }
+thiserror = { version = "1.0", default-features = false }
+tokio = { version = "1.15", default-features = false, features = ["rt", "time"] }
+uuid = { version = "0.8", default-features = false, features = ["v4", "serde"] }
+
+[target.'cfg(all(target_arch = "wasm32", not(target_os = "wasi")))'.dependencies]
+libp2p = { version = "*", features = ["wasm-bindgen", "wasm-ext", "wasm-ext-websocket"] }
+uuid = { version = "*", features = ["wasm-bindgen"] }
+wasm-bindgen-futures = { version = "0.4.29", default-features = false }
+
+[dev-dependencies]
+criterion = { version = "0.3", default-features = false, features = ["stable"] }
+pretty_env_logger = { version = "0.4", default-features = false }
+
+[[bench]]
+name = "benchmark"
+harness = false
+
+[features]
+account = ["identity-account/mem-client", "identity-account/async"]
+default = []

identity-actor/benches/benchmark.rs

@@ -0,0 +1,151 @@
+// Copyright 2020-2022 IOTA Stiftung
+// SPDX-License-Identifier: Apache-2.0
+
+use criterion::criterion_group;
+use criterion::criterion_main;
+use criterion::BenchmarkId;
+use criterion::Criterion;
+use identity_actor::actor::System;
+use identity_actor::actor::SystemBuilder;
+use identity_actor::Multiaddr;
+use identity_actor::PeerId;
+
+use actor::IdentityCreate;
+use actor::RemoteAccount;
+
+async fn setup() -> (System, PeerId, System) {
+  let addr: Multiaddr = "/ip4/0.0.0.0/tcp/0".parse().unwrap();
+  let mut builder = SystemBuilder::new();
+
+  let remote_account = RemoteAccount::new().unwrap();
+  builder.attach::<IdentityCreate, _>(remote_account);
+
+  let mut receiver: System = builder.build().await.unwrap();
+
+  let addr = receiver.start_listening(addr).await.unwrap();
+  let receiver_peer_id = receiver.peer_id();
+
+  let mut sender: System = SystemBuilder::new().build().await.unwrap();
+
+  sender.add_address(receiver_peer_id, addr).await.unwrap();
+
+  (receiver, receiver_peer_id, sender)
+}
+
+fn bench_create_remote_account(c: &mut Criterion) {
+  static ITERATIONS: &[usize] = &[100, 10_000, 100_000];
+
+  let runtime = tokio::runtime::Builder::new_multi_thread()
+    .enable_all()
+    .build()
+    .unwrap();
+
+  let (receiver, receiver_peer_id, sender) = runtime.block_on(setup());
+
+  let mut group = c.benchmark_group("create remote account");
+
+  for size in ITERATIONS.iter() {
+    group.bench_function(BenchmarkId::from_parameter(size), |bencher| {
+      bencher.to_async(&runtime).iter(|| {
+        let mut sender_clone: System = sender.clone();
+
+        async move {
+          sender_clone
+            .send_request(receiver_peer_id, IdentityCreate::default())
+            .await
+            .unwrap()
+            .unwrap();
+        }
+      });
+    });
+  }
+
+  group.finish();
+
+  runtime.block_on(async move {
+    sender.shutdown().await.unwrap();
+    receiver.shutdown().await.unwrap();
+  });
+}
+
+criterion_group!(benches, bench_create_remote_account);
+
+criterion_main!(benches);
+
+mod actor {
+  use dashmap::DashMap;
+  use identity_account::account::Account;
+  use identity_account::account::AccountBuilder;
+  use identity_account::types::IdentitySetup;
+  use identity_actor::actor::Actor;
+  use identity_actor::actor::ActorRequest;
+  use identity_actor::actor::Endpoint;
+  use identity_actor::actor::RequestContext;
+  use identity_iota_core::did::IotaDID;
+  use identity_iota_core::document::IotaDocument;
+  use serde::Deserialize;
+  use serde::Serialize;
+  use std::sync::Arc;
+  use tokio::sync::Mutex;
+
+  #[derive(Debug, Clone)]
+  pub struct RemoteAccount {
+    builder: Arc<Mutex<AccountBuilder>>,
+    accounts: Arc<DashMap<IotaDID, Account>>,
+  }
+
+  impl RemoteAccount {
+    pub fn new() -> identity_account::Result<Self> {
+      let builder: AccountBuilder = Account::builder().autopublish(false);
+
+      Ok(Self {
+        builder: Arc::new(Mutex::new(builder)),
+        accounts: Arc::new(DashMap::new()),
+      })
+    }
+  }
+
+  /// Can be sent to a `RemoteAccount` to instruct it to create an identity.
+  #[derive(Debug, Default, Clone, Serialize, Deserialize)]
+  pub struct IdentityCreate;
+
+  impl From<IdentityCreate> for IdentitySetup {
+    fn from(_: IdentityCreate) -> Self {
+      IdentitySetup::default()
+    }
+  }
+
+  impl ActorRequest for IdentityCreate {
+    type Response = Result<IotaDocument, RemoteAccountError>;
+
+    fn endpoint() -> Endpoint {
+      "remote_account/create".try_into().unwrap()
+    }
+  }
+
+  #[async_trait::async_trait]
+  impl Actor<IdentityCreate> for RemoteAccount {
+    async fn handle(&self, request: RequestContext<IdentityCreate>) -> Result<IotaDocument, RemoteAccountError> {
+      let account: Account = self.builder.lock().await.create_identity(request.input.into()).await?;
+      let doc = account.document().to_owned();
+      self.accounts.insert(account.did().to_owned(), account);
+      Ok(doc)
+    }
+  }
+
+  /// The error type for the [`RemoteAccount`].
+  #[derive(Debug, thiserror::Error, serde::Serialize, serde::Deserialize)]
+  #[non_exhaustive]
+  pub enum RemoteAccountError {
+    #[error("identity not found")]
+    IdentityNotFound,
+    #[error("{0}")]
+    AccountError(String),
+  }
+
+  impl From<identity_account::Error> for RemoteAccountError {
+    fn from(err: identity_account::Error) -> Self {
+      Self::AccountError(err.to_string())
+    }
+  }
+}

identity-actor/src/actor/actor.rs

@@ -0,0 +1,106 @@
+// Copyright 2020-2022 IOTA Stiftung
+// SPDX-License-Identifier: Apache-2.0
+
+use std::fmt::Debug;
+use std::future::Future;
+use std::marker::PhantomData;
+use std::pin::Pin;
+
+use super::ActorRequest;
+use super::RemoteSendError;
+use super::RequestContext;
+use crate::actor::ErrorLocation;
+
+/// A boxed future that is `Send`.
+pub(crate) type BoxFuture<'me, T> = Pin<Box<dyn Future<Output = T> + Send + 'me>>;
+
+/// Actors are objects that encapsulate state and behavior.
+///
+/// Actors handle one or more requests by implementing this trait one or more times
+/// for different `ActorRequest` types.
+///
+/// The requests for an actor are handled synchronously, meaning that the caller waits for
+/// the actor to return its result before continuing.
+#[async_trait::async_trait]
+pub trait Actor<REQ: ActorRequest>: Debug + 'static {
+  /// Called when the system receives a request of type `REQ`.
+  /// The result will be returned to the calling peer.
+  async fn handle(&self, request: RequestContext<REQ>) -> REQ::Response;
+}
+
+/// A trait that wraps a synchronous actor implementation and erases its type.
+/// This allows holding actors with different concrete types in the same collection.
+pub(crate) trait AbstractActor: Debug + Send + Sync + 'static {
+  fn handle(&self, request: RequestContext<Vec<u8>>) -> BoxFuture<'_, Result<Vec<u8>, RemoteSendError>>;
+}
+
+/// A wrapper around synchronous actor implementations that is used for
+/// type erasure together with [`AbstractSyncActor`].
+#[derive(Debug)]
+pub(crate) struct ActorWrapper<ACT, REQ>
+where
+  REQ: ActorRequest + Send + Sync,
+  ACT: Actor<REQ> + Send + Sync,
+{
+  actor: ACT,
+  _phantom_req: PhantomData<REQ>,
+}
+
+impl<ACT, REQ> ActorWrapper<ACT, REQ>
+where
+  REQ: ActorRequest + Send + Sync,
+  ACT: Actor<REQ> + Send + Sync,
+{
+  pub(crate) fn new(actor: ACT) -> Self {
+    Self {
+      actor,
+      _phantom_req: PhantomData,
+    }
+  }
+}
+
+impl<ACT, REQ> AbstractActor for ActorWrapper<ACT, REQ>
+where
+  REQ: ActorRequest + Send + Sync,
+  REQ::Response: Send,
+  ACT: Actor<REQ> + Send + Sync,
+{
+  fn handle(&self, request: RequestContext<Vec<u8>>) -> BoxFuture<'_, Result<Vec<u8>, RemoteSendError>> {
+    let future = async move {
+      let req: REQ =
+        serde_json::from_slice(&request.input).map_err(|error| RemoteSendError::DeserializationFailure {
+          location: ErrorLocation::Remote,
+          context: format!(
+            "deserializing the received bytes into the handler's expected type `{}`",
+            std::any::type_name::<REQ>()
+          ),
+          error_message: error.to_string(),
+        })?;
+
+      let req: RequestContext<REQ> = request.convert(req);
+      let result: REQ::Response = self.actor.handle(req).await;
+      serialize_response::<REQ>(&result)
+    };
+
+    Box::pin(future)
+  }
+}
+
+#[inline(always)]
+fn serialize_response<REQ: ActorRequest>(input: &REQ::Response) -> Result<Vec<u8>, RemoteSendError> {
+  log::debug!(
+    "Attempt serialization into {:?}",
+    std::any::type_name::<REQ::Response>()
+  );
+
+  let response: Vec<u8> = serde_json::to_vec(&input).map_err(|error| RemoteSendError::SerializationFailure {
+    location: ErrorLocation::Remote,
+    context: format!(
+      "serializing the handler's response into `{}`",
+      std::any::type_name::<REQ::Response>()
+    ),
+    error_message: error.to_string(),
+  })?;
+
+  Ok(response)
+}

identity-actor/src/actor/config.rs

@@ -0,0 +1,18 @@
+// Copyright 2020-2022 IOTA Stiftung
+// SPDX-License-Identifier: Apache-2.0
+
+use std::time::Duration;
+
+/// Configuration options for an `Actor`.
+#[derive(Debug, Clone)]
+pub(crate) struct ActorConfig {
+  pub(crate) timeout: Duration,
+}
+
+impl Default for ActorConfig {
+  fn default() -> Self {
+    Self {
+      timeout: Duration::from_secs(30),
+    }
+  }
+}