Repository: tangle-network/gadget (language: rust)

Issue:
[CHECKLIST] MP-ECDSA Protocol
Code from the `dkg-gadget` crate of the `dkg-substrate` repo must be refactored inside the new gadget/protocol model provided by this repository to create a more robust, maintainable, and streamlined implementation.

Since the dkg has two types of protocols (keygen and signing), we can run two different protocols via `webb_gadget::run_protocol` selected twice, sharing some state that allows the signing to use the keygen's output public key. Thus, this task will include these two different phases.

**Phase I: Keygen Protocol**
- [x] #27
- [x] #28

**Phase II: Signing Protocol**
- [x] #29
- [x] #30

Ideally, there would be no need for an in-memory shared state between the `KeygenProtocol` and the `SigningProtocol`. Instead, we would use the off-chain storage as we currently do in the DKG to fetch keys for a specific session.

Since the function `webb_gadget::run_protocol` also requires a connection to a blockchain and a network implementation, we will also need to complete the following tasks:

**Phase III: Networking**
- [x] #31
- [x] #32
- [x] #39

Note: the client should be identical between the keygen and signing protocols. The network will be nearly identical, the only difference being that the underlying network stream will be multiplexed to split signing and keygen messages.

Next, the two protocols for keygen and signing will need to be put together as such:

```rust
let keygen_protocol_future = webb_gadget::run_protocol(keygen_gossip_network, keygen_protocol, blockchain_connection.clone());
let signing_protocol_future = webb_gadget::run_protocol(signing_gossip_network, signing_protocol, blockchain_connection);

tokio::select! {
    res0 = keygen_protocol_future => res0,
    res1 = signing_protocol_future => res1,
}
```

Thus, this step of tying together the two protocols and creating a program entrypoint with a main function will need to be encapsulated in an additional step:

**Phase IV: Application Binary**
- [x] #33

Finally, everything must be tested:

**Phase V: Testing**
- [x] #34
- [x] #35


Discussion / hints:

Fix patch:
diff --git a/.github/workflows/validate_pr.yml b/.github/workflows/validate_pr.yml
index 41b4319de..ed3fb4e63 100644
--- a/.github/workflows/validate_pr.yml
+++ b/.github/workflows/validate_pr.yml
@@ -9,21 +9,6 @@ concurrency:
   cancel-in-progress: true
 
 jobs:
-  integration-tests:
-    name: Integration Tests
-    runs-on: ubuntu-latest
-    steps:
-      - name: Checkout code
-        uses: actions/checkout@v2
-
-      - name: Install protobuf
-        run: sudo apt-get install protobuf-compiler
-
-      - name: Generate certs
-        run: bash ./playground/generate_certs.sh
-
-      - name: Run ZK MVP
-        run: docker-compose up --abort-on-container-exit
   formatting:
     name: Rustfmt
     runs-on: ubuntu-latest
diff --git a/Cargo.toml b/Cargo.toml
index a2f4d2a6f..23c5833a3 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -2,24 +2,61 @@
 resolver = "2"
 members = [
     "gadget-core",
-    "webb-gadget",
+    "gadget-common",
     "zk-gadget",
     "test-gadget",
-    "playground"
+    "protocols/mp-ecdsa",
 ]
 
 [workspace.dependencies]
 gadget-core = { path = "./gadget-core" }
-webb-gadget = { path = "./webb-gadget" }
+gadget-common = { path = "./gadget-common" }
 zk-gadget = { path = "./zk-gadget" }
-zk-playground = { path = "./playground" }
+mp-ecdsa-protocol = { path = "./protocols/mp-ecdsa" }
 
-sc-client-api = { git = "https://github.com/paritytech/substrate", branch = "polkadot-v1.0.0", default-features = false }
-sp-core = { git = "https://github.com/paritytech/substrate", branch = "polkadot-v1.0.0", default-features = false }
-sp-runtime = { git = "https://github.com/paritytech/substrate", branch = "polkadot-v1.0.0", default-features = false }
-sc-utils = { git = "https://github.com/paritytech/substrate", branch = "polkadot-v1.0.0", default-features = false }
+pallet-jobs-rpc-runtime-api = { git = "https://github.com/webb-tools/tangle" }
+tangle-primitives = { git = "https://github.com/webb-tools/tangle" }
+tangle-testnet-runtime = { git = "https://github.com/webb-tools/tangle" }
+tangle-mainnet-runtime = { git = "https://github.com/webb-tools/tangle" }
+
+multi-party-ecdsa = { git = "https://github.com/webb-tools/cggmp-threshold-ecdsa/" }
+round-based = { git = "https://github.com/webb-tools/round-based-protocol", features = [] }
+curv = { package = "curv-kzen", version = "0.10.0" }
+
+sc-client-api = { git = "https://github.com/paritytech/polkadot-sdk", branch = "release-polkadot-v1.1.0" }
+sp-core = { git = "https://github.com/paritytech/polkadot-sdk", branch = "release-polkadot-v1.1.0"  }
+sp-runtime = { git = "https://github.com/paritytech/polkadot-sdk", branch = "release-polkadot-v1.1.0"  }
+sc-utils = { git = "https://github.com/paritytech/polkadot-sdk", branch = "release-polkadot-v1.1.0" }
+sp-api = { git = "https://github.com/paritytech/polkadot-sdk", branch = "release-polkadot-v1.1.0"  }
+sp-application-crypto = { git = "https://github.com/paritytech/polkadot-sdk", branch = "release-polkadot-v1.1.0"  }
+sp-consensus-aura = { git = "https://github.com/paritytech/polkadot-sdk", branch = "release-polkadot-v1.1.0"  }
+sp-keyring = { git = "https://github.com/paritytech/polkadot-sdk", branch = "release-polkadot-v1.1.0"  }
+sp-timestamp = { git = "https://github.com/paritytech/polkadot-sdk", branch = "release-polkadot-v1.1.0"  }
+sp-blockchain = { git = "https://github.com/paritytech/polkadot-sdk", branch = "release-polkadot-v1.1.0"  }
+sp-block-builder = { git = "https://github.com/paritytech/polkadot-sdk", branch = "release-polkadot-v1.1.0" }
+
+sc-offchain = { git = "https://github.com/paritytech/polkadot-sdk", branch = "release-polkadot-v1.1.0"  }
+sc-basic-authorship = { git = "https://github.com/paritytech/polkadot-sdk", branch = "release-polkadot-v1.1.0"  }
+sc-consensus-aura = { git = "https://github.com/paritytech/polkadot-sdk", branch = "release-polkadot-v1.1.0"  }
+sc-consensus-grandpa = { git = "https://github.com/paritytech/polkadot-sdk", branch = "release-polkadot-v1.1.0"  }
+sc-executor = { git = "https://github.com/paritytech/polkadot-sdk", branch = "release-polkadot-v1.1.0"  }
+sc-network = { git = "https://github.com/paritytech/polkadot-sdk", branch = "release-polkadot-v1.1.0"  }
+sc-network-common = { git = "https://github.com/paritytech/polkadot-sdk", branch = "release-polkadot-v1.1.0"  }
+sc-network-sync = { git = "https://github.com/paritytech/polkadot-sdk", branch = "release-polkadot-v1.1.0"  }
+sc-service = { git = "https://github.com/paritytech/polkadot-sdk", branch = "release-polkadot-v1.1.0"  }
+sc-telemetry = { git = "https://github.com/paritytech/polkadot-sdk", branch = "release-polkadot-v1.1.0"  }
+sc-transaction-pool-api = { git = "https://github.com/paritytech/polkadot-sdk", branch = "release-polkadot-v1.1.0"  }
+sc-cli = { git = "https://github.com/paritytech/polkadot-sdk", branch = "release-polkadot-v1.1.0"  }
+sc-consensus = { git = "https://github.com/paritytech/polkadot-sdk", branch = "release-polkadot-v1.1.0" }
+sc-transaction-pool = { git = "https://github.com/paritytech/polkadot-sdk", branch = "release-polkadot-v1.1.0"  }
+sc-rpc-api = { git = "https://github.com/paritytech/polkadot-sdk", branch = "release-polkadot-v1.1.0"  }
 parity-scale-codec = "3.6.5"
 
+substrate-build-script-utils = { git = "https://github.com/paritytech/polkadot-sdk", branch = "release-polkadot-v1.1.0" }
+pallet-im-online = { git = "https://github.com/paritytech/polkadot-sdk", branch = "release-polkadot-v1.1.0"  }
+substrate-frame-rpc-system = { git = "https://github.com/paritytech/polkadot-sdk", branch = "release-polkadot-v1.1.0"  }
+pallet-transaction-payment-rpc = { git = "https://github.com/paritytech/polkadot-sdk", branch = "release-polkadot-v1.1.0"  }
+
 mpc-net = { git = "https://github.com/webb-tools/zk-SaaS/" }
 dist-primitives = { git = "https://github.com/webb-tools/zk-SaaS/" }
 secret-sharing = { git = "https://github.com/webb-tools/zk-SaaS/" }
@@ -38,3 +75,10 @@ hex = "0.4.3"
 uuid = "1.4.1"
 bytes = "1.5.0"
 serde_bytes = "0.11.12"
+itertools = "0.12.0"
+strum = "0.25.0"
+clap = "4.0.32"
+hex-literal = "0.4.1"
+rand = "0.8.5"
+jsonrpsee = "0.16.2"
+linked-hash-map = "0.5.6"
\ No newline at end of file
diff --git a/webb-gadget/Cargo.toml b/gadget-common/Cargo.toml
similarity index 57%
rename from webb-gadget/Cargo.toml
rename to gadget-common/Cargo.toml
index 757b65d95..9d1bc634b 100644
--- a/webb-gadget/Cargo.toml
+++ b/gadget-common/Cargo.toml
@@ -1,25 +1,35 @@
 [package]
-name = "webb-gadget"
+name = "gadget-common"
 version = "0.1.0"
 edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [features]
+default = [
+    "std"
+]
+
+std = [
+    "sp-api/std",
+    "tangle-primitives/std",
+    "pallet-jobs-rpc-runtime-api/std"
+]
 
 [dependencies]
-mpc-net = { workspace = true }
-tokio-rustls = { workspace = true }
 gadget-core = { workspace = true, features = ["substrate"] }
 tokio = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
-tokio-util = { version = "0.7.9" }
 async-trait = { workspace = true }
 log = { workspace = true }
 parking_lot = { workspace = true }
 futures-util = { workspace = true }
-auto_impl = "1.1.0"
 sc-client-api = { workspace = true }
 sp-core = { workspace = true }
 sp-runtime = { workspace = true }
-hex = { workspace = true }
\ No newline at end of file
+hex = { workspace = true }
+bincode2 = { workspace = true }
+
+sp-api = { workspace = true, default-features = false }
+tangle-primitives = { workspace = true, default-features = false }
+pallet-jobs-rpc-runtime-api = { workspace = true, default-features = false }
\ No newline at end of file
diff --git a/gadget-common/src/client.rs b/gadget-common/src/client.rs
new file mode 100644
index 000000000..989bc4f38
--- /dev/null
+++ b/gadget-common/src/client.rs
@@ -0,0 +1,146 @@
+use crate::debug_logger::DebugLogger;
+use crate::keystore::{ECDSAKeyStore, KeystoreBackend};
+use async_trait::async_trait;
+use gadget_core::gadget::substrate::Client;
+use pallet_jobs_rpc_runtime_api::{JobsApi, RpcResponseJobsData};
+use sc_client_api::{
+    Backend, BlockImportNotification, BlockchainEvents, FinalityNotification, HeaderBackend,
+};
+use sp_api::BlockT as Block;
+use sp_api::ProvideRuntimeApi;
+use std::error::Error;
+use std::sync::Arc;
+use tangle_primitives::jobs::{JobId, JobKey, JobResult};
+
+pub struct MpEcdsaClient<B: Block, BE, KBE: KeystoreBackend, C> {
+    client: Arc<C>,
+    pub key_store: ECDSAKeyStore<KBE>,
+    logger: DebugLogger,
+    _block: std::marker::PhantomData<(BE, B)>,
+}
+
+impl<B: Block, BE, KBE: KeystoreBackend, C> Clone for MpEcdsaClient<B, BE, KBE, C> {
+    fn clone(&self) -> Self {
+        Self {
+            client: self.client.clone(),
+            key_store: self.key_store.clone(),
+            logger: self.logger.clone(),
+            _block: self._block,
+        }
+    }
+}
+
+pub async fn create_client<
+    B: Block,
+    BE: Backend<B>,
+    KBE: KeystoreBackend,
+    C: ClientWithApi<B, BE>,
+>(
+    client: Arc<C>,
+    logger: DebugLogger,
+    key_store: ECDSAKeyStore<KBE>,
+) -> Result<MpEcdsaClient<B, BE, KBE, C>, Box<dyn Error>>
+where
+    <C as ProvideRuntimeApi<B>>::Api: JobsApi<B, AccountId>,
+{
+    Ok(MpEcdsaClient {
+        client,
+        key_store,
+        logger,
+        _block: std::marker::PhantomData,
+    })
+}
+
+pub type AccountId = sp_core::ecdsa::Public;
+
+pub trait ClientWithApi<B, BE>:
+    BlockchainEvents<B> + HeaderBackend<B> + ProvideRuntimeApi<B> + Send + Sync + Client<B> + 'static
+where
+    B: Block,
+    BE: Backend<B>,
+    <Self as ProvideRuntimeApi<B>>::Api: JobsApi<B, AccountId>,
+{
+}
+
+impl<B, BE, T> ClientWithApi<B, BE> for T
+where
+    B: Block,
+    BE: Backend<B>,
+    T: BlockchainEvents<B>
+        + HeaderBackend<B>
+        + ProvideRuntimeApi<B>
+        + Send
+        + Sync
+        + Client<B>
+        + 'static,
+    <T as ProvideRuntimeApi<B>>::Api: JobsApi<B, AccountId>,
+{
+}
+
+impl<B: Block, BE: Backend<B>, KBE: KeystoreBackend, C: ClientWithApi<B, BE>>
+    MpEcdsaClient<B, BE, KBE, C>
+where
+    <C as ProvideRuntimeApi<B>>::Api: JobsApi<B, AccountId>,
+{
+    pub async fn query_jobs_by_validator(
+        &self,
+        at: <B as Block>::Hash,
+        validator: AccountId,
+    ) -> Result<Vec<RpcResponseJobsData<AccountId>>, crate::Error> {
+        exec_client_function(&self.client, move |client| {
+            client.runtime_api().query_jobs_by_validator(at, validator)
+        })
+        .await
+        .map_err(|err| crate::Error::ClientError {
+            err: format!("Failed to query jobs by validator: {err:?}"),
+        })?
+        .map_err(|err| crate::Error::ClientError {
+            err: format!("Failed to query jobs by validator: {err:?}"),
+        })
+    }
+
+    // TODO: Proper implementation
+    pub async fn submit_job_result(
+        &self,
+        _job_key: JobKey,
+        _job_id: JobId,
+        _result: JobResult,
+    ) -> Result<(), crate::Error> {
+        Ok(())
+    }
+}
+
+#[async_trait]
+impl<B, BE, KBE, C> Client<B> for MpEcdsaClient<B, BE, KBE, C>
+where
+    B: Block,
+    BE: Backend<B>,
+    C: ClientWithApi<B, BE>,
+    KBE: KeystoreBackend,
+    <C as ProvideRuntimeApi<B>>::Api: JobsApi<B, AccountId>,
+{
+    async fn get_next_finality_notification(&self) -> Option<FinalityNotification<B>> {
+        self.client.get_next_finality_notification().await
+    }
+
+    async fn get_latest_finality_notification(&self) -> Option<FinalityNotification<B>> {
+        self.client.get_latest_finality_notification().await
+    }
+
+    async fn get_next_block_import_notification(&self) -> Option<BlockImportNotification<B>> {
+        self.client.get_next_block_import_notification().await
+    }
+}
+
+async fn exec_client_function<C, F, T>(client: &Arc<C>, function: F) -> T
+where
+    for<'a> F: FnOnce(&'a C) -> T,
+    C: Send + Sync + 'static,
+    T: Send + 'static,
+    F: Send + 'static,
+{
+    let client = client.clone();
+    tokio::task::spawn_blocking(move || function(&client))
+        .await
+        .expect("Failed to spawn blocking task")
+}
diff --git a/gadget-common/src/debug_logger.rs b/gadget-common/src/debug_logger.rs
new file mode 100644
index 000000000..43d09d243
--- /dev/null
+++ b/gadget-common/src/debug_logger.rs
@@ -0,0 +1,28 @@
+use std::fmt::Display;
+
+#[derive(Clone)]
+pub struct DebugLogger {
+    pub peer_id: String,
+}
+
+impl DebugLogger {
+    pub fn trace<T: Display>(&self, msg: T) {
+        log::trace!(target: "gadget", "[{}] {msg}", &self.peer_id);
+    }
+
+    pub fn debug<T: Display>(&self, msg: T) {
+        log::debug!(target: "gadget", "[{}] {msg}", &self.peer_id);
+    }
+
+    pub fn info<T: Display>(&self, msg: T) {
+        log::info!(target: "gadget", "[{}] {msg}", &self.peer_id);
+    }
+
+    pub fn warn<T: Display>(&self, msg: T) {
+        log::warn!(target: "gadget", "[{}] {msg}", &self.peer_id);
+    }
+
+    pub fn error<T: Display>(&self, msg: T) {
+        log::error!(target: "gadget", "[{}] {msg}", &self.peer_id);
+    }
+}
diff --git a/webb-gadget/src/gadget/message.rs b/gadget-common/src/gadget/message.rs
similarity index 53%
rename from webb-gadget/src/gadget/message.rs
rename to gadget-common/src/gadget/message.rs
index c018cf23a..13ae69ba0 100644
--- a/webb-gadget/src/gadget/message.rs
+++ b/gadget-common/src/gadget/message.rs
@@ -2,17 +2,21 @@ use crate::gadget::work_manager::WebbWorkManager;
 use gadget_core::job_manager::{ProtocolMessageMetadata, WorkManagerInterface};
 use serde::{Deserialize, Serialize};
 
-#[derive(Serialize, Deserialize, Clone)]
+#[derive(Serialize, Deserialize, Clone, Debug)]
 pub struct GadgetProtocolMessage {
     pub associated_block_id: <WebbWorkManager as WorkManagerInterface>::Clock,
     pub associated_session_id: <WebbWorkManager as WorkManagerInterface>::SessionID,
-    pub associated_ssid: <WebbWorkManager as WorkManagerInterface>::SSID,
+    pub associated_retry_id: <WebbWorkManager as WorkManagerInterface>::RetryID,
     // A unique marker for the associated task this message belongs to
     pub task_hash: <WebbWorkManager as WorkManagerInterface>::TaskID,
     pub from: UserID,
     // If None, this is a broadcasted message
     pub to: Option<UserID>,
     pub payload: Vec<u8>,
+    // Some protocols need this additional data
+    pub from_account_id: Option<sp_core::ecdsa::Public>,
+    // Some protocol need this additional data
+    pub to_account_id: Option<sp_core::ecdsa::Public>,
 }
 
 pub type UserID = u32;
@@ -26,11 +30,29 @@ impl ProtocolMessageMetadata<WebbWorkManager> for GadgetProtocolMessage {
         self.associated_session_id
     }
 
-    fn associated_ssid(&self) -> <WebbWorkManager as WorkManagerInterface>::SSID {
-        self.associated_ssid
+    fn associated_retry_id(&self) -> <WebbWorkManager as WorkManagerInterface>::RetryID {
+        self.associated_retry_id
     }
 
     fn associated_task(&self) -> <WebbWorkManager as WorkManagerInterface>::TaskID {
         self.task_hash
     }
+
+    fn associated_sender_user_id(&self) -> <WebbWorkManager as WorkManagerInterface>::UserID {
+        self.from
+    }
+
+    fn associated_recipient_user_id(
+        &self,
+    ) -> Option<<WebbWorkManager as WorkManagerInterface>::UserID> {
+        self.to
+    }
+}
+
+impl GadgetProtocolMessage {
+    /// Returns a hash of the message
+    pub fn hash(&self) -> Vec<u8> {
+        let serialized = bincode2::serialize(&self).expect("Should serialize");
+        sp_core::keccak_256(&serialized).to_vec()
+    }
 }
diff --git a/webb-gadget/src/gadget/mod.rs b/gadget-common/src/gadget/mod.rs
similarity index 66%
rename from webb-gadget/src/gadget/mod.rs
rename to gadget-common/src/gadget/mod.rs
index 339dd2787..0d4ff4024 100644
--- a/webb-gadget/src/gadget/mod.rs
+++ b/gadget-common/src/gadget/mod.rs
@@ -13,6 +13,7 @@ use sp_runtime::traits::{Block, Header};
 use sp_runtime::SaturatedConversion;
 use std::marker::PhantomData;
 use std::sync::Arc;
+use std::time::Duration;
 
 pub mod message;
 pub mod network;
@@ -27,23 +28,20 @@ pub struct WebbModule<B, C, N, M> {
     _pd: PhantomData<(B, C)>,
 }
 
-const MAX_ACTIVE_TASKS: usize = 4;
-const MAX_PENDING_TASKS: usize = 4;
+const DEFAULT_MAX_ACTIVE_TASKS: usize = 4;
+const DEFAULT_MAX_PENDING_TASKS: usize = 4;
+const DEFAULT_POLL_INTERVAL: Option<Duration> = Some(Duration::from_millis(200));
 
-impl<C: Client<B>, B: Block, N: Network, M: WebbGadgetProtocol<B>> WebbModule<B, C, N, M> {
-    pub fn new(network: N, module: M, now: Option<u64>) -> Self {
-        let clock = Arc::new(RwLock::new(now));
-        let clock_clone = clock.clone();
-
-        let job_manager_zk = WebbWorkManager::new(move || *clock_clone.read());
-
-        let job_manager = ProtocolWorkManager::new(
-            job_manager_zk,
-            MAX_ACTIVE_TASKS,
-            MAX_PENDING_TASKS,
-            PollMethod::Interval { millis: 200 },
-        );
+#[derive(Debug)]
+pub struct WorkManagerConfig {
+    pub interval: Option<Duration>,
+    pub max_active_tasks: usize,
+    pub max_pending_tasks: usize,
+}
 
+impl<C: Client<B>, B: Block, N: Network, M: WebbGadgetProtocol<B>> WebbModule<B, C, N, M> {
+    pub fn new(network: N, module: M, job_manager: ProtocolWorkManager<WebbWorkManager>) -> Self {
+        let clock = job_manager.utility.clock.clone();
         WebbModule {
             protocol: module,
             job_manager,
@@ -74,22 +72,30 @@ impl<C: Client<B>, B: Block, N: Network, M: WebbGadgetProtocol<B>> SubstrateGadg
         let now: u64 = (*notification.header.number()).saturated_into();
         *self.clock.write() = Some(now);
 
-        while let Some(job) = self
+        if let Some(jobs) = self
             .protocol
-            .get_next_job(&notification, now, &self.job_manager)
+            .get_next_jobs(&notification, now, &self.job_manager)
             .await?
         {
-            let (remote, protocol) = job;
-            let task_id = remote.associated_task_id;
-
-            if let Err(err) = self
-                .job_manager
-                .push_task(task_id, false, Arc::new(remote), protocol)
-            {
-                log::error!("Failed to push task to job manager: {err:?}");
+            for job in jobs {
+                let (remote, protocol) = job;
+                let task_id = remote.associated_task_id;
+
+                if let Err(err) =
+                    self.job_manager
+                        .push_task(task_id, false, Arc::new(remote), protocol)
+                {
+                    log::error!("Failed to push task to job manager: {err:?}");
+                }
             }
         }
 
+        // Poll jobs on each finality notification if we're using manual polling.
+        // This helps synchronize the actions of nodes in the network
+        if self.job_manager.poll_method() == PollMethod::Manual {
+            self.job_manager.poll();
+        }
+
         Ok(())
     }
 
@@ -121,16 +127,23 @@ pub type Job = (AsyncProtocolRemote, BuiltExecutableJobWrapper);
 
 #[async_trait]
 pub trait WebbGadgetProtocol<B: Block>: Send + Sync {
-    async fn get_next_job(
+    async fn get_next_jobs(
         &self,
         notification: &FinalityNotification<B>,
         now: u64,
         job_manager: &ProtocolWorkManager<WebbWorkManager>,
-    ) -> Result<Option<Job>, Error>;
+    ) -> Result<Option<Vec<Job>>, Error>;
     async fn process_block_import_notification(
         &self,
         notification: BlockImportNotification<B>,
         job_manager: &ProtocolWorkManager<WebbWorkManager>,
     ) -> Result<(), Error>;
     async fn process_error(&self, error: Error, job_manager: &ProtocolWorkManager<WebbWorkManager>);
+    fn get_work_manager_config(&self) -> WorkManagerConfig {
+        WorkManagerConfig {
+            interval: DEFAULT_POLL_INTERVAL,
+            max_active_tasks: DEFAULT_MAX_ACTIVE_TASKS,
+            max_pending_tasks: DEFAULT_MAX_PENDING_TASKS,
+        }
+    }
 }
diff --git a/webb-gadget/src/gadget/network.rs b/gadget-common/src/gadget/network.rs
similarity index 90%
rename from webb-gadget/src/gadget/network.rs
rename to gadget-common/src/gadget/network.rs
index ec73b80ce..4c2003909 100644
--- a/webb-gadget/src/gadget/network.rs
+++ b/gadget-common/src/gadget/network.rs
@@ -4,7 +4,7 @@ use async_trait::async_trait;
 use gadget_core::job_manager::WorkManagerInterface;
 
 #[async_trait]
-pub trait Network: Send + Sync + Clone {
+pub trait Network: Send + Sync + Clone + 'static {
     async fn next_message(
         &self,
     ) -> Option<<WebbWorkManager as WorkManagerInterface>::ProtocolMessage>;
diff --git a/webb-gadget/src/gadget/work_manager.rs b/gadget-common/src/gadget/work_manager.rs
similarity index 57%
rename from webb-gadget/src/gadget/work_manager.rs
rename to gadget-common/src/gadget/work_manager.rs
index a9fc188a0..9aea888ec 100644
--- a/webb-gadget/src/gadget/work_manager.rs
+++ b/gadget-common/src/gadget/work_manager.rs
@@ -1,33 +1,17 @@
-use crate::gadget::message::GadgetProtocolMessage;
+use crate::gadget::message::{GadgetProtocolMessage, UserID};
 use gadget_core::job_manager::WorkManagerInterface;
+use parking_lot::RwLock;
 use std::sync::Arc;
 
 pub struct WebbWorkManager {
-    pub(crate) clock: Arc<
-        dyn Fn() -> Option<<WebbWorkManager as WorkManagerInterface>::Clock>
-            + Send
-            + Sync
-            + 'static,
-    >,
-}
-
-impl WebbWorkManager {
-    pub fn new(
-        clock: impl Fn() -> Option<<WebbWorkManager as WorkManagerInterface>::Clock>
-            + Send
-            + Sync
-            + 'static,
-    ) -> Self {
-        Self {
-            clock: Arc::new(clock),
-        }
-    }
+    pub(crate) clock: Arc<RwLock<Option<<WebbWorkManager as WorkManagerInterface>::Clock>>>,
 }
 
 const ACCEPTABLE_BLOCK_TOLERANCE: u64 = 5;
 
 impl WorkManagerInterface for WebbWorkManager {
-    type SSID = u16;
+    type RetryID = u16;
+    type UserID = UserID;
     type Clock = u64;
     type ProtocolMessage = GadgetProtocolMessage;
     type Error = crate::Error;
@@ -47,7 +31,9 @@ impl WorkManagerInterface for WebbWorkManager {
     }
 
     fn clock(&self) -> Self::Clock {
-        (self.clock)().expect("No finality notification received")
+        self.clock
+            .read()
+            .expect("No finality notification received")
     }
 
     fn acceptable_block_tolerance() -> Self::Clock {
diff --git a/webb-gadget/src/helpers.rs b/gadget-common/src/helpers.rs
similarity index 97%
rename from webb-gadget/src/helpers.rs
rename to gadget-common/src/helpers.rs
index 9f1e0cd38..589c63b7e 100644
--- a/webb-gadget/src/helpers.rs
+++ b/gadget-common/src/helpers.rs
@@ -62,6 +62,7 @@ pub fn create_job_manager_compatible_job<T: Display + Send + Clone + 'static>(
 
     JobBuilder::default()
         .pre(pre_hook)
+        .protocol(wrapped_future)
         .post(post_hook)
-        .build(wrapped_future)
+        .build()
 }
diff --git a/gadget-common/src/keystore.rs b/gadget-common/src/keystore.rs
new file mode 100644
index 000000000..54d9e4e01
--- /dev/null
+++ b/gadget-common/src/keystore.rs
@@ -0,0 +1,83 @@
+use async_trait::async_trait;
+use parking_lot::RwLock;
+use serde::de::DeserializeOwned;
+use serde::Serialize;
+use std::collections::HashMap;
+use std::sync::Arc;
+use tangle_primitives::jobs::JobId;
+
+#[derive(Clone)]
+pub struct ECDSAKeyStore<BE: KeystoreBackend> {
+    backend: BE,
+}
+
+impl<BE: KeystoreBackend> ECDSAKeyStore<BE> {
+    pub fn in_memory() -> ECDSAKeyStore<InMemoryBackend> {
+        ECDSAKeyStore {
+            backend: InMemoryBackend::new(),
+        }
+    }
+
+    pub async fn get<T: DeserializeOwned>(
+        &self,
+        job_id: &JobId,
+    ) -> Result<Option<T>, crate::Error> {
+        self.backend.get(job_id).await
+    }
+
+    pub async fn set<T: Serialize + Send>(
+        &self,
+        job_id: JobId,
+        value: T,
+    ) -> Result<(), crate::Error> {
+        self.backend.set(job_id, value).await
+    }
+}
+
+#[async_trait]
+pub trait KeystoreBackend: Clone + Send + Sync + 'static {
+    async fn get<T: DeserializeOwned>(&self, job_id: &JobId) -> Result<Option<T>, crate::Error>;
+    async fn set<T: Serialize + Send>(&self, job_id: JobId, value: T) -> Result<(), crate::Error>;
+}
+
+#[derive(Clone)]
+pub struct InMemoryBackend {
+    map: Arc<RwLock<HashMap<JobId, Vec<u8>>>>,
+}
+
+impl Default for InMemoryBackend {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl InMemoryBackend {
+    pub fn new() -> Self {
+        Self {
+            map: Arc::new(RwLock::new(HashMap::new())),
+        }
+    }
+}
+
+#[async_trait]
+impl KeystoreBackend for InMemoryBackend {
+    async fn get<T: DeserializeOwned>(&self, job_id: &JobId) -> Result<Option<T>, crate::Error> {
+        if let Some(bytes) = self.map.read().get(job_id).cloned() {
+            let value: T =
+                bincode2::deserialize(&bytes).map_err(|rr| crate::Error::KeystoreError {
+                    err: format!("Failed to deserialize value: {:?}", rr),
+                })?;
+            Ok(Some(value))
+        } else {
+            Ok(None)
+        }
+    }
+
+    async fn set<T: Serialize + Send>(&self, job_id: JobId, value: T) -> Result<(), crate::Error> {
+        let serialized = bincode2::serialize(&value).map_err(|rr| crate::Error::KeystoreError {
+            err: format!("Failed to serialize value: {:?}", rr),
+        })?;
+        self.map.write().insert(job_id, serialized);
+        Ok(())
+    }
+}
diff --git a/webb-gadget/src/lib.rs b/gadget-common/src/lib.rs
similarity index 56%
rename from webb-gadget/src/lib.rs
rename to gadget-common/src/lib.rs
index 3e857020d..6275c1b88 100644
--- a/webb-gadget/src/lib.rs
+++ b/gadget-common/src/lib.rs
@@ -1,18 +1,24 @@
 use crate::gadget::network::Network;
+use crate::gadget::work_manager::WebbWorkManager;
 use crate::gadget::{WebbGadgetProtocol, WebbModule};
 use futures_util::future::TryFutureExt;
 use gadget_core::gadget::manager::{GadgetError, GadgetManager};
 use gadget_core::gadget::substrate::{Client, SubstrateGadget};
-use gadget_core::job_manager::WorkManagerError;
+use gadget_core::job::JobError;
+use gadget_core::job_manager::{PollMethod, ProtocolWorkManager, WorkManagerError};
+use parking_lot::RwLock;
 pub use sc_client_api::BlockImportNotification;
 pub use sc_client_api::{Backend, FinalityNotification};
-use sp_runtime::traits::{Block, Header};
+pub use sp_runtime::traits::{Block, Header};
 use sp_runtime::SaturatedConversion;
 use std::fmt::{Debug, Display, Formatter};
+use std::sync::Arc;
 
+pub mod client;
+pub mod debug_logger;
 pub mod gadget;
-
 pub mod helpers;
+pub mod keystore;
 pub mod protocol;
 
 #[derive(Debug)]
@@ -27,6 +33,9 @@ pub enum Error {
     WorkManagerError { err: WorkManagerError },
     ProtocolRemoteError { err: String },
     ClientError { err: String },
+    JobError { err: JobError },
+    NetworkError { err: String },
+    KeystoreError { err: String },
 }
 
 impl Display for Error {
@@ -37,11 +46,43 @@ impl Display for Error {
 
 impl std::error::Error for Error {}
 
+impl From<JobError> for Error {
+    fn from(err: JobError) -> Self {
+        Error::JobError { err }
+    }
+}
+
+pub async fn run_protocol_with_wm<C: Client<B>, B: Block, N: Network, P: WebbGadgetProtocol<B>>(
+    network: N,
+    protocol: P,
+    client: C,
+    work_manager: ProtocolWorkManager<WebbWorkManager>,
+) -> Result<(), Error> {
+    let webb_module = WebbModule::new(network.clone(), protocol, work_manager);
+    // Plug the module into the substrate gadget to interface the WebbGadget with Substrate
+    let substrate_gadget = SubstrateGadget::new(client, webb_module);
+
+    let network_future = network.run();
+    let gadget_future =
+        GadgetManager::new(substrate_gadget).map_err(|err| Error::GadgetManagerError { err });
+
+    // Run both the network and the gadget together
+    tokio::try_join!(network_future, gadget_future).map(|_| ())
+}
+
 pub async fn run_protocol<C: Client<B>, B: Block, N: Network, P: WebbGadgetProtocol<B>>(
     network: N,
     protocol: P,
     client: C,
 ) -> Result<(), Error> {
+    let wm = create_work_manager(&client, &protocol).await?;
+    run_protocol_with_wm(network, protocol, client, wm).await
+}
+
+pub async fn create_work_manager<C: Client<B>, B: Block, P: WebbGadgetProtocol<B>>(
+    client: &C,
+    protocol: &P,
+) -> Result<ProtocolWorkManager<WebbWorkManager>, Error> {
     // Before running, wait for the first finality notification we receive
     let now: u64 = (*client
         .get_latest_finality_notification()
@@ -53,14 +94,23 @@ pub async fn run_protocol<C: Client<B>, B: Block, N: Network, P: WebbGadgetProto
         .number())
     .saturated_into();
 
-    let webb_module = WebbModule::new(network.clone(), protocol, Some(now));
-    // Plug the module into the substrate gadget to interface the WebbGadget with Substrate
-    let substrate_gadget = SubstrateGadget::new(client, webb_module);
+    let work_manager_config = protocol.get_work_manager_config();
 
-    let network_future = network.run();
-    let gadget_future =
-        GadgetManager::new(substrate_gadget).map_err(|err| Error::GadgetManagerError { err });
+    let clock = Arc::new(RwLock::new(Some(now)));
 
-    // Run both the network and the gadget together
-    tokio::try_join!(network_future, gadget_future).map(|_| ())
+    let job_manager_zk = WebbWorkManager { clock };
+
+    let poll_method = match work_manager_config.interval {
+        Some(interval) => PollMethod::Interval {
+            millis: interval.as_millis() as u64,
+        },
+        None => PollMethod::Manual,
+    };
+
+    Ok(ProtocolWorkManager::new(
+        job_manager_zk,
+        work_manager_config.max_active_tasks,
+        work_manager_config.max_pending_tasks,
+        poll_method,
+    ))
 }
diff --git a/webb-gadget/src/protocol.rs b/gadget-common/src/protocol.rs
similarity index 84%
rename from webb-gadget/src/protocol.rs
rename to gadget-common/src/protocol.rs
index 5092e3f4e..db7bf461e 100644
--- a/webb-gadget/src/protocol.rs
+++ b/gadget-common/src/protocol.rs
@@ -1,5 +1,6 @@
 use crate::gadget::message::GadgetProtocolMessage;
 use crate::gadget::work_manager::WebbWorkManager;
+use crate::gadget::Job;
 use async_trait::async_trait;
 use gadget_core::job::{BuiltExecutableJobWrapper, JobError};
 use gadget_core::job_manager::{ProtocolRemote, ShutdownReason, WorkManagerInterface};
@@ -13,7 +14,7 @@ pub struct AsyncProtocolRemote {
     pub shutdown_tx: Mutex<Option<tokio::sync::oneshot::Sender<ShutdownReason>>>,
     pub associated_session_id: <WebbWorkManager as WorkManagerInterface>::SessionID,
     pub associated_block_id: <WebbWorkManager as WorkManagerInterface>::Clock,
-    pub associated_ssid: <WebbWorkManager as WorkManagerInterface>::SSID,
+    pub associated_retry_id: <WebbWorkManager as WorkManagerInterface>::RetryID,
     pub associated_task_id: <WebbWorkManager as WorkManagerInterface>::TaskID,
     pub to_async_protocol: tokio::sync::mpsc::UnboundedSender<
         <WebbWorkManager as WorkManagerInterface>::ProtocolMessage,
@@ -23,22 +24,25 @@ pub struct AsyncProtocolRemote {
 
 #[async_trait]
 pub trait AsyncProtocol {
+    type AdditionalParams: Send + Sync + 'static;
     async fn generate_protocol_from(
         &self,
         associated_block_id: <WebbWorkManager as WorkManagerInterface>::Clock,
-        associated_ssid: <WebbWorkManager as WorkManagerInterface>::SSID,
+        associated_retry_id: <WebbWorkManager as WorkManagerInterface>::RetryID,
         associated_session_id: <WebbWorkManager as WorkManagerInterface>::SessionID,
         associated_task_id: <WebbWorkManager as WorkManagerInterface>::TaskID,
         protocol_message_rx: UnboundedReceiver<GadgetProtocolMessage>,
+        additional_params: Self::AdditionalParams,
     ) -> Result<BuiltExecutableJobWrapper, JobError>;
 
     async fn create(
         &self,
         session_id: <WebbWorkManager as WorkManagerInterface>::SessionID,
         now: <WebbWorkManager as WorkManagerInterface>::Clock,
-        ssid: <WebbWorkManager as WorkManagerInterface>::SSID,
+        retry_id: <WebbWorkManager as WorkManagerInterface>::RetryID,
         task_id: <WebbWorkManager as WorkManagerInterface>::TaskID,
-    ) -> Result<(AsyncProtocolRemote, BuiltExecutableJobWrapper), JobError> {
+        additional_params: Self::AdditionalParams,
+    ) -> Result<Job, JobError> {
         let is_done = Arc::new(AtomicBool::new(false));
         let (to_async_protocol, protocol_message_rx) = tokio::sync::mpsc::unbounded_channel();
         let (start_tx, start_rx) = tokio::sync::oneshot::channel();
@@ -46,14 +50,21 @@ pub trait AsyncProtocol {
 
         let proto_hash_hex = hex::encode(task_id);
         let async_protocol = self
-            .generate_protocol_from(now, ssid, session_id, task_id, protocol_message_rx)
+            .generate_protocol_from(
+                now,
+                retry_id,
+                session_id,
+                task_id,
+                protocol_message_rx,
+                additional_params,
+            )
             .await?;
 
         let remote = AsyncProtocolRemote {
             start_tx: Mutex::new(Some(start_tx)),
             shutdown_tx: Mutex::new(Some(shutdown_tx)),
             associated_block_id: now,
-            associated_ssid: ssid,
+            associated_retry_id: retry_id,
             associated_task_id: task_id,
             associated_session_id: session_id,
             to_async_protocol,
@@ -90,8 +101,6 @@ impl ProtocolRemote<WebbWorkManager> for AsyncProtocolRemote {
         self.associated_session_id
     }
 
-    fn set_as_primary(&self) {}
-
     fn started_at(&self) -> <WebbWorkManager as WorkManagerInterface>::Clock {
         self.associated_block_id
     }
@@ -131,7 +140,7 @@ impl ProtocolRemote<WebbWorkManager> for AsyncProtocolRemote {
         self.start_tx.lock().is_none()
     }
 
-    fn ssid(&self) -> <WebbWorkManager as WorkManagerInterface>::SSID {
-        self.associated_ssid
+    fn retry_id(&self) -> <WebbWorkManager as WorkManagerInterface>::RetryID {
+        self.associated_retry_id
     }
 }
diff --git a/gadget-core/Cargo.toml b/gadget-core/Cargo.toml
index f4479ffb4..790dd5db5 100644
--- a/gadget-core/Cargo.toml
+++ b/gadget-core/Cargo.toml
@@ -9,18 +9,18 @@ edition = "2021"
 substrate = [
     "sp-runtime",
     "sc-client-api",
-    "futures"
 ]
 
 [dependencies]
 parking_lot = { workspace = true }
 tokio = { workspace = true, features = ["sync", "time", "macros", "rt"] }
 hex = "0.4.3"
+log = { workspace = true }
 async-trait = "0.1.73"
+auto_impl = "1.1.0"
 
-sp-runtime = { optional = true, workspace = true, default-features = false }
-sc-client-api = { optional = true, workspace = true, default-features = false }
-futures = { optional = true, workspace = true }
+sp-runtime = { optional = true, workspace = true }
+sc-client-api = { optional = true, workspace = true }
 
 [dev-dependencies]
 tokio = { workspace = true, features = ["macros"]}
\ No newline at end of file
diff --git a/gadget-core/src/gadget/substrate/mod.rs b/gadget-core/src/gadget/substrate/mod.rs
index 2d1324d79..de7e23ed5 100644
--- a/gadget-core/src/gadget/substrate/mod.rs
+++ b/gadget-core/src/gadget/substrate/mod.rs
@@ -1,5 +1,6 @@
 use crate::gadget::manager::AbstractGadget;
 use async_trait::async_trait;
+use auto_impl::auto_impl;
 use sc_client_api::{BlockImportNotification, FinalityNotification};
 use sp_runtime::traits::Block;
 use std::error::Error;
@@ -47,6 +48,7 @@ impl Display for SubstrateGadgetError {
 impl Error for SubstrateGadgetError {}
 
 #[async_trait]
+#[auto_impl(Arc)]
 pub trait Client<B: Block>: Send + Sync {
     async fn get_next_finality_notification(&self) -> Option<FinalityNotification<B>>;
     async fn get_latest_finality_notification(&self) -> Option<FinalityNotification<B>>;
diff --git a/gadget-core/src/job.rs b/gadget-core/src/job.rs
index eaad4eb5b..4bbebc9e3 100644
--- a/gadget-core/src/job.rs
+++ b/gadget-core/src/job.rs
@@ -95,11 +95,13 @@ where
 #[derive(Default)]
 pub struct JobBuilder {
     pre: Option<Pin<Box<PreJobHook>>>,
+    protocol: Option<Pin<Box<ProtocolJobHook>>>,
     post: Option<Pin<Box<PostJobHook>>>,
 }
 
 pub type PreJobHook = dyn SendFuture<'static, Result<ProceedWithExecution, JobError>>;
 pub type PostJobHook = dyn SendFuture<'static, Result<(), JobError>>;
+pub type ProtocolJobHook = dyn SendFuture<'static, Result<(), JobError>>;
 
 pub struct DefaultPreJobHook;
 impl Future for DefaultPreJobHook {
@@ -138,6 +140,14 @@ impl JobBuilder {
         self
     }
 
+    pub fn protocol<Protocol>(mut self, protocol: Protocol) -> Self
+    where
+        Protocol: SendFuture<'static, Result<(), JobError>>,
+    {
+        self.protocol = Some(Box::pin(protocol));
+        self
+    }
+
     pub fn post<Post>(mut self, post: Post) -> Self
     where
         Post: SendFuture<'static, Result<(), JobError>>,
@@ -146,10 +156,7 @@ impl JobBuilder {
         self
     }
 
-    pub fn build<Protocol>(self, protocol: Protocol) -> BuiltExecutableJobWrapper
-    where
-        Protocol: SendFuture<'static, Result<(), JobError>>,
-    {
+    pub fn build(self) -> BuiltExecutableJobWrapper {
         let pre = if let Some(pre) = self.pre {
             pre
         } else {
@@ -162,9 +169,11 @@ impl JobBuilder {
             Box::pin(DefaultPostJobHook)
         };
 
+        let protocol = Box::pin(self.protocol.expect("Must specify protocol"));
+
         ExecutableJobWrapper {
             pre,
-            protocol: Box::pin(protocol),
+            protocol,
             post,
         }
     }
@@ -235,21 +244,20 @@ mod tests {
         let counter_clone2 = counter.clone();
         let counter_final = counter.clone();
 
-        let protocol = async move {
-            counter_clone.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
-            Ok(())
-        };
-
         let mut job = super::JobBuilder::new()
             .pre(async move {
                 counter.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                 Ok(super::ProceedWithExecution::True)
             })
+            .protocol(async move {
+                counter_clone.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+                Ok(())
+            })
             .post(async move {
                 counter_clone2.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                 Ok(())
             })
-            .build(protocol);
+            .build();
 
         job.execute().await.unwrap();
         assert_eq!(counter_final.load(std::sync::atomic::Ordering::SeqCst), 3);
@@ -262,17 +270,16 @@ mod tests {
         let counter_clone2 = counter.clone();
         let counter_final = counter.clone();
 
-        let protocol = async move {
-            counter_clone.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
-            Ok(())
-        };
-
         let mut job = super::JobBuilder::default()
+            .protocol(async move {
+                counter_clone.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+                Ok(())
+            })
             .post(async move {
                 counter_clone2.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                 Ok(())
             })
-            .build(protocol);
+            .build();
 
         job.execute().await.unwrap();
         assert_eq!(counter_final.load(std::sync::atomic::Ordering::SeqCst), 2);
@@ -284,17 +291,16 @@ mod tests {
         let counter_clone = counter.clone();
         let counter_final = counter.clone();
 
-        let protocol = async move {
-            counter_clone.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
-            Ok(())
-        };
-
         let mut job = super::JobBuilder::default()
             .pre(async move {
                 counter.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                 Ok(super::ProceedWithExecution::True)
             })
-            .build(protocol);
+            .protocol(async move {
+                counter_clone.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+                Ok(())
+            })
+            .build();
 
         job.execute().await.unwrap();
         assert_eq!(counter_final.load(std::sync::atomic::Ordering::SeqCst), 2);
@@ -306,12 +312,12 @@ mod tests {
         let counter_clone = counter.clone();
         let counter_final = counter.clone();
 
-        let protocol = async move {
-            counter_clone.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
-            Ok(())
-        };
-
-        let mut job = super::JobBuilder::default().build(protocol);
+        let mut job = super::JobBuilder::default()
+            .protocol(async move {
+                counter_clone.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+                Ok(())
+            })
+            .build();
 
         job.execute().await.unwrap();
         assert_eq!(counter_final.load(std::sync::atomic::Ordering::SeqCst), 1);
diff --git a/gadget-core/src/job_manager.rs b/gadget-core/src/job_manager.rs
index b12ac56f5..29b583776 100644
--- a/gadget-core/src/job_manager.rs
+++ b/gadget-core/src/job_manager.rs
@@ -24,7 +24,7 @@ pub enum DeliveryType {
 }
 pub struct ProtocolWorkManager<WM: WorkManagerInterface> {
     inner: Arc<RwLock<WorkManagerInner<WM>>>,
-    utility: Arc<WM>,
+    pub utility: Arc<WM>,
     // for now, use a hard-coded value for the number of tasks
     max_tasks: Arc<usize>,
     max_enqueued_tasks: Arc<usize>,
@@ -47,13 +47,15 @@ pub struct WorkManagerInner<WM: WorkManagerInterface> {
     pub active_tasks: HashSet<Job<WM>>,
     pub enqueued_tasks: VecDeque<Job<WM>>,
     // task hash => SSID => enqueued messages
-    pub enqueued_messages: EnqueuedMessage<WM::TaskID, WM::SSID, WM::ProtocolMessage>,
+    pub enqueued_messages: EnqueuedMessage<WM::TaskID, WM::RetryID, WM::ProtocolMessage>,
+    pub retry_id_tracker: HashMap<WM::TaskID, WM::RetryID>,
 }
 
 pub type EnqueuedMessage<A, B, C> = HashMap<A, HashMap<B, VecDeque<C>>>;
 
 pub trait WorkManagerInterface: Send + Sync + 'static + Sized {
-    type SSID: Copy + Hash + Eq + PartialEq + Send + Sync + 'static;
+    type RetryID: Copy + Hash + Eq + PartialEq + Send + Sync + 'static;
+    type UserID: Copy + Hash + Eq + PartialEq + Send + Sync + 'static;
     type Clock: Copy
         + Debug
         + Default
@@ -99,8 +101,10 @@ fn saturating_sub<T: Sub<Output = T> + Ord + Default>(a: T, b: T) -> T {
 pub trait ProtocolMessageMetadata<WM: WorkManagerInterface> {
     fn associated_block_id(&self) -> WM::Clock;
     fn associated_session_id(&self) -> WM::SessionID;
-    fn associated_ssid(&self) -> WM::SSID;
+    fn associated_retry_id(&self) -> WM::RetryID;
     fn associated_task(&self) -> WM::TaskID;
+    fn associated_sender_user_id(&self) -> WM::UserID;
+    fn associated_recipient_user_id(&self) -> Option<WM::UserID>;
 }
 
 /// The [`ProtocolRemote`] is the interface between the [`ProtocolWorkManager`] and the async protocol.
@@ -108,13 +112,12 @@ pub trait ProtocolMessageMetadata<WM: WorkManagerInterface> {
 pub trait ProtocolRemote<WM: WorkManagerInterface>: Send + Sync + 'static {
     fn start(&self) -> Result<(), WM::Error>;
     fn session_id(&self) -> WM::SessionID;
-    fn set_as_primary(&self);
     fn started_at(&self) -> WM::Clock;
     fn shutdown(&self, reason: ShutdownReason) -> Result<(), WM::Error>;
     fn is_done(&self) -> bool;
     fn deliver_message(&self, message: WM::ProtocolMessage) -> Result<(), WM::Error>;
     fn has_started(&self) -> bool;
-    fn ssid(&self) -> WM::SSID;
+    fn retry_id(&self) -> WM::RetryID;
 
     fn has_stalled(&self, now: WM::Clock) -> bool {
         now >= self.started_at() + WM::acceptable_block_tolerance()
@@ -147,6 +150,7 @@ impl<WM: WorkManagerInterface> ProtocolWorkManager<WM> {
                 active_tasks: HashSet::new(),
                 enqueued_tasks: VecDeque::new(),
                 enqueued_messages: HashMap::new(),
+                retry_id_tracker: HashMap::new(),
             })),
             utility: Arc::new(utility),
             max_tasks: Arc::new(max_tasks),
@@ -192,8 +196,6 @@ impl<WM: WorkManagerInterface> ProtocolWorkManager<WM> {
         task: T,
     ) -> Result<(), WorkManagerError> {
         let mut lock = self.inner.write();
-        // set as primary, that way on drop, the async protocol ends
-        handle.set_as_primary();
         let job = Job {
             task: Arc::new(parking_lot::Mutex::new(Some(Box::new(task)))),
             handle,
@@ -306,7 +308,7 @@ impl<WM: WorkManagerInterface> ProtocolWorkManager<WM> {
         // Next, remove any outdated enqueued messages to prevent RAM bloat
         let mut to_remove = vec![];
         for (hash, queue) in lock.enqueued_messages.iter_mut() {
-            for (ssid, queue) in queue.iter_mut() {
+            for (retry_id, queue) in queue.iter_mut() {
                 let before = queue.len();
                 // Only keep the messages that are not outdated
                 queue.retain(|msg| {
@@ -323,17 +325,17 @@ impl<WM: WorkManagerInterface> ProtocolWorkManager<WM> {
                 }
 
                 if queue.is_empty() {
-                    to_remove.push((*hash, *ssid));
+                    to_remove.push((*hash, *retry_id));
                 }
             }
         }
 
         // Next, to prevent the existence of piling-up empty *inner* queues, remove them
-        for (hash, ssid) in to_remove {
+        for (hash, retry_id) in to_remove {
             lock.enqueued_messages
                 .get_mut(&hash)
                 .expect("Should be available")
-                .remove(&ssid);
+                .remove(&retry_id);
         }
 
         // Finally, remove any empty outer maps
@@ -359,8 +361,8 @@ impl<WM: WorkManagerInterface> ProtocolWorkManager<WM> {
         } else {
             // deliver all the enqueued messages to the protocol now
             if let Some(mut enqueued_messages_map) = lock.enqueued_messages.remove(&job.task_hash) {
-                let job_ssid = job.handle.ssid();
-                if let Some(mut enqueued_messages) = enqueued_messages_map.remove(&job_ssid) {
+                let job_retry_id = job.handle.retry_id();
+                if let Some(mut enqueued_messages) = enqueued_messages_map.remove(&job_retry_id) {
                     self.utility.debug(format!(
                         "Will now deliver {} enqueued message(s) to the async protocol for {:?}",
                         enqueued_messages.len(),
@@ -391,15 +393,34 @@ impl<WM: WorkManagerInterface> ProtocolWorkManager<WM> {
         let task = job.task.clone();
         // Put the job inside here, that way the drop code does not get called right away,
         // killing the process
+        let task_hash = job.task_hash;
+        *lock
+            .retry_id_tracker
+            .entry(task_hash)
+            .or_insert(job.handle.retry_id()) = job.handle.retry_id();
         lock.active_tasks.insert(job);
+
         let mut task = task.lock().take().expect("Should not happen");
-        let task = async move { task.execute().await };
+        let on_task_finish_handle = self.inner.clone();
+        let task = async move {
+            if task.execute().await.is_ok() {
+                on_task_finish_handle
+                    .write()
+                    .retry_id_tracker
+                    .remove(&task_hash);
+            }
+        };
 
         // Spawn the task. When it finishes, it will clean itself up
         tokio::task::spawn(task);
         Ok(())
     }
 
+    pub fn latest_retry_id(&self, task_hash: &WM::TaskID) -> Option<WM::RetryID> {
+        let lock = self.inner.read();
+        lock.retry_id_tracker.get(task_hash).copied()
+    }
+
     pub fn job_exists(&self, job: &WM::TaskID) -> bool {
         let lock = self.inner.read();
         lock.active_tasks.iter().any(|r| &r.task_hash == job)
@@ -414,6 +435,17 @@ impl<WM: WorkManagerInterface> ProtocolWorkManager<WM> {
             "Delivered message is intended for session_id = {}",
             msg.associated_session_id()
         ));
+
+        if let Some(recv) = msg.associated_recipient_user_id() {
+            if recv == msg.associated_sender_user_id() {
+                self.utility
+                    .warn("Message sent to self, which is not allowed".to_string());
+                return Err(WorkManagerError::DeliverMessageFailed {
+                    reason: "Message sent to self, which is not allowed".to_string(),
+                });
+            }
+        }
+
         let message_task_hash = msg.associated_task();
         let mut lock = self.inner.write();
 
@@ -468,12 +500,16 @@ impl<WM: WorkManagerInterface> ProtocolWorkManager<WM> {
         lock.enqueued_messages
             .entry(message_task_hash)
             .or_default()
-            .entry(msg.associated_ssid())
+            .entry(msg.associated_retry_id())
             .or_default()
             .push_back(msg);
 
         Ok(DeliveryType::EnqueuedMessage)
     }
+
+    pub fn poll_method(&self) -> PollMethod {
+        *self.poll_method
+    }
 }
 
 pub struct Job<WM: WorkManagerInterface> {
@@ -541,7 +577,7 @@ fn should_deliver<WM: WorkManagerInterface>(
 ) -> bool {
     task.handle.session_id() == msg.associated_session_id()
         && task.task_hash == message_task_hash
-        && task.handle.ssid() == msg.associated_ssid()
+        && task.handle.retry_id() == msg.associated_retry_id()
         && WM::associated_block_id_acceptable(
             task.handle.started_at(), // use to be associated_block_id
             msg.associated_block_id(),
@@ -551,7 +587,7 @@ fn should_deliver<WM: WorkManagerInterface>(
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::job::{BuiltExecutableJobWrapper, JobBuilder};
+    use crate::job::{BuiltExecutableJobWrapper, JobBuilder, JobError};
     use parking_lot::Mutex;
     use std::sync::atomic::{AtomicBool, Ordering};
     use std::sync::Arc;
@@ -565,8 +601,10 @@ mod tests {
         message: String,
         associated_block_id: u64,
         associated_session_id: u32,
-        associated_ssid: u32,
+        associated_retry_id: u32,
         associated_task: [u8; 32],
+        associated_sender: u64,
+        associated_recipient: Option<u64>,
     }
 
     impl ProtocolMessageMetadata<TestWorkManager> for TestMessage {
@@ -576,28 +614,41 @@ mod tests {
         fn associated_session_id(&self) -> u32 {
             self.associated_session_id
         }
-        fn associated_ssid(&self) -> u32 {
-            self.associated_ssid
+        fn associated_retry_id(&self) -> u32 {
+            self.associated_retry_id
         }
         fn associated_task(&self) -> [u8; 32] {
             self.associated_task
         }
+
+        fn associated_sender_user_id(&self) -> <TestWorkManager as WorkManagerInterface>::UserID {
+            self.associated_sender
+        }
+
+        fn associated_recipient_user_id(
+            &self,
+        ) -> Option<<TestWorkManager as WorkManagerInterface>::UserID> {
+            self.associated_recipient
+        }
     }
 
     impl WorkManagerInterface for TestWorkManager {
-        type SSID = u32;
+        type RetryID = u32;
+        type UserID = u64;
         type Clock = u64;
         type ProtocolMessage = TestMessage;
         type Error = ();
         type SessionID = u32;
         type TaskID = [u8; 32];
 
-        fn debug(&self, _input: String) {}
+        fn debug(&self, input: String) {
+            log::debug!("{input}")
+        }
         fn error(&self, input: String) {
-            println!("ERROR: {input}")
+            log::error!("{input}")
         }
         fn warn(&self, input: String) {
-            println!("WARN: {input}")
+            log::warn!("{input}")
         }
         fn clock(&self) -> Self::Clock {
             0
@@ -611,7 +662,7 @@ mod tests {
     #[derive(Clone)]
     pub struct TestProtocolRemote {
         session_id: u32,
-        ssid: u32,
+        retry_id: u32,
         started_at: u64,
         delivered_messages: UnboundedSender<TestMessage>,
         start_tx: Arc<Mutex<Option<tokio::sync::oneshot::Sender<()>>>>,
@@ -622,7 +673,7 @@ mod tests {
     impl TestProtocolRemote {
         pub fn new(
             session_id: u32,
-            ssid: u32,
+            retry_id: u32,
             started_at: u64,
             task_tx: UnboundedSender<TestMessage>,
             start_tx: tokio::sync::oneshot::Sender<()>,
@@ -630,7 +681,7 @@ mod tests {
         ) -> Arc<Self> {
             Arc::new(Self {
                 session_id,
-                ssid,
+                retry_id,
                 started_at,
                 delivered_messages: task_tx,
                 start_tx: Arc::new(Mutex::new(Some(start_tx))),
@@ -643,7 +694,7 @@ mod tests {
     #[allow(clippy::type_complexity)]
     pub fn generate_async_protocol(
         session_id: u32,
-        ssid: u32,
+        retry_id: u32,
         started_at: u64,
     ) -> (
         Arc<TestProtocolRemote>,
@@ -653,14 +704,52 @@ mod tests {
         let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
         let (start_tx, start_rx) = tokio::sync::oneshot::channel();
         let is_done = Arc::new(AtomicBool::new(false));
-        let remote =
-            TestProtocolRemote::new(session_id, ssid, started_at, tx, start_tx, is_done.clone());
+        let remote = TestProtocolRemote::new(
+            session_id,
+            retry_id,
+            started_at,
+            tx,
+            start_tx,
+            is_done.clone(),
+        );
         let protocol = async move {
             start_rx.await.unwrap();
             is_done.store(true, Ordering::SeqCst);
             Ok(())
         };
-        let task = JobBuilder::new().build(protocol);
+        let task = JobBuilder::new().protocol(protocol).build();
+        (remote, task, rx)
+    }
+
+    #[allow(clippy::type_complexity)]
+    pub fn generate_async_protocol_error(
+        session_id: u32,
+        retry_id: u32,
+        started_at: u64,
+    ) -> (
+        Arc<TestProtocolRemote>,
+        BuiltExecutableJobWrapper,
+        UnboundedReceiver<TestMessage>,
+    ) {
+        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
+        let (start_tx, start_rx) = tokio::sync::oneshot::channel();
+        let is_done = Arc::new(AtomicBool::new(false));
+        let remote = TestProtocolRemote::new(
+            session_id,
+            retry_id,
+            started_at,
+            tx,
+            start_tx,
+            is_done.clone(),
+        );
+        let protocol = async move {
+            start_rx.await.unwrap();
+            is_done.store(true, Ordering::SeqCst);
+            Err(JobError {
+                reason: "Intentional error".to_string(),
+            })
+        };
+        let task = JobBuilder::new().protocol(protocol).build();
         (remote, task, rx)
     }
 
@@ -673,7 +762,6 @@ mod tests {
         fn session_id(&self) -> u32 {
             self.session_id
         }
-        fn set_as_primary(&self) {}
         fn has_stalled(&self, now: u64) -> bool {
             now > self.started_at
         }
@@ -695,8 +783,8 @@ mod tests {
         fn is_active(&self) -> bool {
             self.has_started() && !self.is_done()
         }
-        fn ssid(&self) -> u32 {
-            self.ssid
+        fn retry_id(&self) -> u32 {
+            self.retry_id
         }
     }
 
@@ -721,8 +809,10 @@ mod tests {
             message: "test".to_string(),
             associated_block_id: 0,
             associated_session_id: 0,
-            associated_ssid: 0,
+            associated_retry_id: 0,
             associated_task: [0; 32],
+            associated_recipient: None,
+            associated_sender: 0,
         };
         assert_ne!(
             DeliveryType::EnqueuedMessage,
@@ -731,6 +821,27 @@ mod tests {
         let _ = rx.recv().await.unwrap();
     }
 
+    #[tokio::test]
+    async fn test_deliver_self_ref_message() {
+        let work_manager = ProtocolWorkManager::new(TestWorkManager, 10, 10, PollMethod::Manual);
+        let (remote, task, _rx) = generate_async_protocol(0, 0, 0);
+
+        work_manager
+            .push_task(Default::default(), true, remote.clone(), task)
+            .unwrap();
+
+        let message = TestMessage {
+            message: "test".to_string(),
+            associated_block_id: 0,
+            associated_session_id: 0,
+            associated_retry_id: 0,
+            associated_task: [0; 32],
+            associated_recipient: Some(0),
+            associated_sender: 0,
+        };
+        assert!(work_manager.deliver_message(message).is_err());
+    }
+
     #[tokio::test]
     async fn test_job_exists() {
         let work_manager = ProtocolWorkManager::new(TestWorkManager, 10, 10, PollMethod::Manual);
@@ -787,8 +898,10 @@ mod tests {
             message: "test".to_string(),
             associated_block_id: 0,
             associated_session_id: 1,
-            associated_ssid: 1,
+            associated_retry_id: 1,
             associated_task: [0; 32],
+            associated_recipient: None,
+            associated_sender: 0,
         };
         assert_ne!(
             DeliveryType::EnqueuedMessage,
@@ -1018,8 +1131,10 @@ mod tests {
             message: "test".to_string(),
             associated_block_id: 0,
             associated_session_id: 1,
-            associated_ssid: 1,
+            associated_retry_id: 1,
             associated_task: [0; 32],
+            associated_sender: 0,
+            associated_recipient: None,
         };
 
         // Deliver a message to a non-existent job
@@ -1042,8 +1157,10 @@ mod tests {
             message: "test".to_string(),
             associated_block_id: 10, // Outdated block ID
             associated_session_id: 1,
-            associated_ssid: 1,
+            associated_retry_id: 1,
             associated_task: [0; 32],
+            associated_sender: 0,
+            associated_recipient: None,
         };
 
         // Try to deliver a message with an outdated block ID
@@ -1078,8 +1195,10 @@ mod tests {
             message: "test".to_string(),
             associated_block_id: 0,
             associated_session_id: 1,
-            associated_ssid: 1,
+            associated_retry_id: 1,
             associated_task: [0; 32],
+            associated_sender: 0,
+            associated_recipient: None,
         };
 
         for _ in 0..10 {
@@ -1139,8 +1258,10 @@ mod tests {
             message: "test".to_string(),
             associated_block_id: 0,
             associated_session_id: 1,
-            associated_ssid: 1,
+            associated_retry_id: 1,
             associated_task: [0; 32],
+            associated_sender: 0,
+            associated_recipient: None,
         };
 
         work_manager.push_task([0; 32], true, remote, task).unwrap();
@@ -1158,8 +1279,10 @@ mod tests {
             message: "test".to_string(),
             associated_block_id: 0,
             associated_session_id: 1,
-            associated_ssid: 1,
+            associated_retry_id: 1,
             associated_task: [1; 32], // incorrect task hash, not 0;32 as below
+            associated_sender: 0,
+            associated_recipient: None,
         };
 
         work_manager.push_task([0; 32], true, remote, task).unwrap();
@@ -1168,6 +1291,40 @@ mod tests {
         assert_eq!(delivery_type, DeliveryType::EnqueuedMessage); // message should be enqueued because the task hash is incorrect
     }
 
+    #[tokio::test]
+    async fn test_retry_id() {
+        let work_manager = ProtocolWorkManager::new(TestWorkManager, 1, 0, PollMethod::Manual);
+        let (remote, task, _rx) = generate_async_protocol_error(1, 0, 0);
+        assert!(work_manager.latest_retry_id(&[0; 32]).is_none());
+        work_manager.push_task([0; 32], true, remote, task).unwrap();
+        tokio::time::sleep(Duration::from_millis(100)).await;
+        assert_eq!(
+            work_manager
+                .latest_retry_id(&[0; 32])
+                .expect("Should exist"),
+            0
+        );
+
+        // Simulate a retry
+        let (remote, task, _rx) = generate_async_protocol_error(1, 1, 0);
+        work_manager.push_task([0; 32], true, remote, task).unwrap();
+        tokio::time::sleep(Duration::from_millis(100)).await;
+
+        assert_eq!(
+            work_manager
+                .latest_retry_id(&[0; 32])
+                .expect("Should exist"),
+            1
+        );
+
+        // Simulate that the protocol works on the third attempt
+        let (remote, task, _rx) = generate_async_protocol(1, 2, 0);
+        work_manager.push_task([0; 32], true, remote, task).unwrap();
+        tokio::time::sleep(Duration::from_millis(100)).await;
+        // Since the protocol was a success, the retry id should be cleared
+        assert!(work_manager.latest_retry_id(&[0; 32]).is_none());
+    }
+
     struct DummyRangeChecker<const N: u64>;
     struct DummyProtocolMessage;
 
@@ -1178,16 +1335,29 @@ mod tests {
         fn associated_session_id(&self) -> u64 {
             0u64
         }
-        fn associated_ssid(&self) -> u64 {
+        fn associated_retry_id(&self) -> u64 {
             0u64
         }
         fn associated_task(&self) -> [u8; 32] {
             [0; 32]
         }
+
+        fn associated_sender_user_id(
+            &self,
+        ) -> <DummyRangeChecker<N> as WorkManagerInterface>::UserID {
+            0
+        }
+
+        fn associated_recipient_user_id(
+            &self,
+        ) -> Option<<DummyRangeChecker<N> as WorkManagerInterface>::UserID> {
+            None
+        }
     }
 
     impl<const N: u64> WorkManagerInterface for DummyRangeChecker<N> {
-        type SSID = u64;
+        type RetryID = u64;
+        type UserID = u64;
         type Clock = u64;
         type ProtocolMessage = DummyProtocolMessage;
         type Error = ();
diff --git a/playground/Cargo.toml b/playground/Cargo.toml
deleted file mode 100644
index a5998a282..000000000
--- a/playground/Cargo.toml
+++ /dev/null
@@ -1,72 +0,0 @@
-[package]
-name = "zk-playground"
-version = "0.1.0"
-authors = ["Shady Khalifa <dev+github@shadykhalifa.me>"]
-license = "GPL-3.0-or-later WITH Classpath-exception-2.0"
-edition = "2021"
-
-[[example]]
-name = "zk-test-client"
-path = "examples/client.rs"
-
-[[example]]
-name = "zknode"
-path = "examples/node.rs"
-
-[[example]]
-name = "certgen"
-path = "examples/certgen.rs"
-
-[features]
-default = ["parallel"]
-parallel = ["ark-std/parallel", "groth16/parallel", "rayon"]
-
-[dependencies]
-clap = { version = "4.4", features = ["derive", "wrap_help"] }
-anyhow = "1"
-sync_wrapper = "0.1.2"
-parking_lot = { workspace = true }
-tokio = { workspace = true, features = ["sync", "time", "macros", "rt"] }
-hex = "0.4.3"
-serde = { workspace = true, features = ["derive"] }
-serde_bytes = { workspace = true }
-serde_json = "1"
-async-trait = "0.1.73"
-rcgen = "0.11.3"
-rustls = "0.21.7"
-tokio-rustls = "0.24.1"
-rustls-pemfile = "1.0.3"
-tracing = "0.1"
-tracing-subscriber = { version = "0.3.17", features = ["env-filter"] }
-rayon = { version = "1.8.0", optional = true }
-rand = { version = "0.8", default-features = false, features = ["std"] }
-
-sc-utils = { workspace = true }
-uuid = { workspace = true, features = ["v4"] }
-sp-runtime = { workspace = true, default-features = false }
-sc-client-api = { workspace = true, default-features = false }
-parity-scale-codec = { workspace = true, features = ["derive"] }
-futures = { workspace = true }
-
-zk-gadget = { workspace = true }
-gadget-core = { workspace = true }
-webb-gadget = { workspace = true }
-mpc-net = { workspace = true }
-groth16 = { workspace = true }
-secret-sharing = { workspace = true }
-
-# ARK Libraries
-ark-std = {version = "0.4.0", default-features = false, features = [ "print-trace", "std" ]}
-ark-crypto-primitives = {version = "0.4.0", default-features = false}
-ark-ff = {version = "0.4.0", default-features = false}
-ark-poly = {version = "0.4.0", default-features = false}
-ark-ec = {version = "0.4.0", default-features = false}
-ark-relations = { version = "0.4.0", default-features = false }
-ark-serialize = { version = "0.4.0", default-features = false, features = [ "derive" ] }
-ark-groth16 = { version = "0.4.0", default-features = false }
-ark-circom = { git = "https://github.com/arkworks-rs/circom-compat.git" }
-
-# ARK curves
-ark-bn254 = {version = "0.4.0", default-features = false, features = ["curve"] }
-
-[dev-dependencies]
diff --git a/playground/README.md b/playground/README.md
deleted file mode 100644
index bd136c55f..000000000
--- a/playground/README.md
+++ /dev/null
@@ -1,101 +0,0 @@
-## How to run this playground
-
-1. Generate the Local Certificates:
-```sh
-./playground/generate_certs.sh
-```
-2. Generate the shares by running the client:
-```sh
-cargo run --release --example zk-test-client -- --circuit-id 1 --job-id 1 --wasm ./fixtures/sha256/circom.wasm --r1cs ./fixtures/sha256/circom.r1cs --input ./fixtures/sha256/input.json --public-inputs ./fixtures/sha256/public_inputs.json --output-dir ./target/zk --generate-proving-key
-```
-You can change the `--circuit-id` and `--job-id` to any number you want. The `--output-dir` is where the shares will be generated. You can change the `--wasm`, `--r1cs`, `--input` and `--public-inputs` to any circuit you want to run.
-Note: `--generate-proving-key` will generate the proving key. If you already have the proving key, you can omit this flag.
-
-3. Running the zknode(s):
-You will need 8 terminals to run the zknode(s). In each terminal, run:
-
-**King**
-
-```sh
-RUST_LOG=debug cargo run --release --example zknode -- --king-ip 127.0.0.1:5555 \
-    --public_identity_der certs/0/cert.der \
-    --watch-dir target/zk \
-    --name king --i 0 \
-    --private-identity-der certs/0/key.der
-```
-
-**Node 1**
-
-```sh
-RUST_LOG=debug cargo run --release --example zknode -- --king-ip 127.0.0.1:5555 \
-    --public_identity_der certs/1/cert.der \
-    --client_only_king_public_identity_der certs/0/cert.der \
-    --watch-dir target/zk \
-    --name node1 --i 1 \
-    --private-identity-der certs/1/key.der
-```
-
-**Node 2**
-```sh
-RUST_LOG=debug cargo run --release --example zknode -- --king-ip 127.0.0.1:5555 \
-    --public_identity_der certs/2/cert.der \
-    --client_only_king_public_identity_der certs/0/cert.der \
-    --watch-dir target/zk \
-    --name node2 --i 2 \
-    --private-identity-der certs/2/key.der
-```
-
-**Node 3**
-```sh
-RUST_LOG=debug cargo run --release --example zknode -- --king-ip 127.0.0.1:5555 \
-    --public_identity_der certs/3/cert.der \
-    --client_only_king_public_identity_der certs/0/cert.der \
-    --watch-dir target/zk \
-    --name node3 --i 3 \
-    --private-identity-der certs/3/key.der
-```
-
-
-**Node 4**
-```sh
-RUST_LOG=debug cargo run --release --example zknode -- --king-ip 127.0.0.1:5555 \
-    --public_identity_der certs/4/cert.der \
-    --client_only_king_public_identity_der certs/0/cert.der \
-    --watch-dir target/zk \
-    --name node4 --i 4 \
-    --private-identity-der certs/4/key.der
-```
-
-**Node 5**
-```sh
-RUST_LOG=debug cargo run --release --example zknode -- --king-ip 127.0.0.1:5555 \
-    --public_identity_der certs/5/cert.der \
-    --client_only_king_public_identity_der certs/0/cert.der \
-    --watch-dir target/zk \
-    --name node5 --i 5 \
-    --private-identity-der certs/5/key.der
-```
-
-**Node 6**
-```sh
-RUST_LOG=debug cargo run --release --example zknode -- --king-ip 127.0.0.1:5555 \
-    --public_identity_der certs/6/cert.der \
-    --client_only_king_public_identity_der certs/0/cert.der \
-    --watch-dir target/zk \
-    --name node6 --i 6 \
-    --private-identity-der certs/6/key.der
-```
-
-**Node 7**
-```sh
-RUST_LOG=debug cargo run --release --example zknode -- --king-ip 127.0.0.1:5555 \
-    --public_identity_der certs/7/cert.der \
-    --client_only_king_public_identity_der certs/0/cert.der \
-    --watch-dir target/zk \
-    --name node7 --i 7 \
-    --private-identity-der certs/7/key.der
-```
-
-4. Wait till the king finishes and outputs the final result.
-
-
diff --git a/playground/examples/certgen.rs b/playground/examples/certgen.rs
deleted file mode 100644
index 0014860d5..000000000
--- a/playground/examples/certgen.rs
+++ /dev/null
@@ -1,44 +0,0 @@
-#![allow(clippy::complexity, clippy::style, clippy::pedantic)]
-
-use clap::Parser;
-use rcgen::{date_time_ymd, Certificate, CertificateParams, DistinguishedName, DnType, SanType};
-use std::fs;
-
-/// ZK Node
-#[derive(Parser, Debug)]
-#[command(author, version, about, long_about = None)]
-struct Args {
-    /// The Id of the node, also acts as the party index
-    #[arg(long)]
-    i: usize,
-}
-
-fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let args = Args::parse();
-    let mut params: CertificateParams = Default::default();
-    params.not_before = date_time_ymd(1975, 01, 01);
-    params.not_after = date_time_ymd(4096, 01, 01);
-    params.distinguished_name = DistinguishedName::new();
-    params
-        .distinguished_name
-        .push(DnType::OrganizationName, "Webb Inc.");
-    params
-        .distinguished_name
-        .push(DnType::CommonName, "Master Cert");
-    params.subject_alt_names = vec![
-        SanType::IpAddress("127.0.0.1".parse()?),
-        SanType::IpAddress("0.0.0.0".parse()?),
-        SanType::IpAddress("172.40.0.10".parse()?), // For docker container
-        SanType::DnsName("localhost".to_string()),
-    ];
-
-    let cert = Certificate::from_params(params)?;
-
-    std::fs::create_dir_all(format!("certs/{}", args.i))?;
-    fs::write(format!("certs/{}/cert.der", args.i), &cert.serialize_der()?)?;
-    fs::write(
-        format!("certs/{}/key.der", args.i),
-        &cert.serialize_private_key_der(),
-    )?;
-    Ok(())
-}
diff --git a/playground/examples/client.rs b/playground/examples/client.rs
deleted file mode 100644
index d21b181db..000000000
--- a/playground/examples/client.rs
+++ /dev/null
@@ -1,222 +0,0 @@
-use std::path::PathBuf;
-
-use ark_bn254::{Bn254, Fr as Bn254Fr};
-use ark_circom::{CircomBuilder, CircomConfig, CircomReduction};
-use ark_crypto_primitives::snark::SNARK;
-use ark_ec::pairing::Pairing;
-use ark_ff::Field;
-use ark_groth16::Groth16;
-use ark_poly::Radix2EvaluationDomain;
-use ark_relations::r1cs::{ConstraintSynthesizer, ConstraintSystem};
-use ark_serialize::CanonicalSerialize;
-use ark_std::{cfg_chunks, cfg_into_iter, end_timer, start_timer, Zero};
-
-use clap::Parser;
-use groth16::qap::qap;
-use serde_bytes::ByteBuf;
-use tracing_subscriber::fmt::SubscriberBuilder;
-use tracing_subscriber::util::SubscriberInitExt;
-use tracing_subscriber::EnvFilter;
-
-use rand::SeedableRng;
-use secret_sharing::pss::PackedSharingParams;
-use zk_gadget::client_ext::job_types::{CircuitProperties, JobProperties, PackedQAPShare};
-
-#[cfg(feature = "parallel")]
-use rayon::prelude::*;
-
-fn pack_from_witness<E: Pairing>(
-    pp: &PackedSharingParams<E::ScalarField>,
-    full_assignment: Vec<E::ScalarField>,
-) -> Vec<Vec<E::ScalarField>> {
-    let packed_assignments = cfg_chunks!(full_assignment, pp.l)
-        .map(|chunk| {
-            let secrets = if chunk.len() < pp.l {
-                let mut secrets = chunk.to_vec();
-                secrets.resize(pp.l, E::ScalarField::zero());
-                secrets
-            } else {
-                chunk.to_vec()
-            };
-            pp.pack_from_public(secrets)
-        })
-        .collect::<Vec<_>>();
-
-    cfg_into_iter!(0..pp.n)
-        .map(|i| {
-            cfg_into_iter!(0..packed_assignments.len())
-                .map(|j| packed_assignments[j][i])
-                .collect::<Vec<_>>()
-        })
-        .collect::<Vec<_>>()
-}
-
-/// Client
-#[derive(Parser, Debug)]
-#[command(author, version, about, long_about = None)]
-struct Args {
-    /// Circuit Id
-    #[arg(long)]
-    circuit_id: u64,
-    /// Job Id
-    #[arg(long)]
-    job_id: u64,
-    /// Circom wasm file
-    #[arg(long)]
-    wasm: PathBuf,
-    /// Circom r1cs file
-    #[arg(long)]
-    r1cs: PathBuf,
-    /// Circuit input as json file
-    #[arg(long)]
-    input: PathBuf,
-    /// Public inputs as json file
-    #[arg(long)]
-    public_inputs: PathBuf,
-    /// Output directory where we write the proving key and the shares.
-    #[arg(long)]
-    output_dir: PathBuf,
-    /// Shall we generate the proving key?
-    #[arg(long)]
-    generate_proving_key: bool,
-}
-
-#[tokio::main]
-async fn main() -> anyhow::Result<()> {
-    setup_log();
-    let args = Args::parse();
-    let setup_timer = start_timer!(|| "Setup");
-    let cfg = CircomConfig::<Bn254>::new(&args.wasm, &args.r1cs).unwrap();
-    let mut builder = CircomBuilder::new(cfg);
-    let rng = &mut ark_std::rand::rngs::StdRng::from_seed([42u8; 32]);
-    let input = tokio::fs::read_to_string(args.input).await?;
-    // input is a json map of string to BigInt values; BigInt is serialized as a string
-    // {"a": 1, "b": 2}
-    let input = serde_json::from_str::<serde_json::Value>(&input)?;
-    let obj = match input.as_object() {
-        Some(obj) => obj,
-        None => anyhow::bail!("Input must be a json map of string to BigInt values"),
-    };
-    for (k, v) in obj {
-        match v {
-            serde_json::Value::Number(u) => builder.push_input(k, u.as_u64().unwrap()),
-            _ => anyhow::bail!("Input must be a json map of string to BigInt values"),
-        }
-    }
-    // read the public inputs
-    // public inputs is a json array of BigInt values; BigInt is serialized as a string
-    let public_inputs = tokio::fs::read_to_string(args.public_inputs).await?;
-    let public_inputs = serde_json::from_str::<serde_json::Value>(&public_inputs)?;
-    let public_inputs = match public_inputs.as_array() {
-        Some(arr) => arr
-            .iter()
-            .map(|v| v.as_str().unwrap().to_owned())
-            .collect::<Vec<_>>(),
-        None => anyhow::bail!("Public inputs must be a json array of BigInt values"),
-    };
-    if args.generate_proving_key {
-        let circuit = builder.setup();
-        let (pk, _vk) = Groth16::<Bn254, CircomReduction>::circuit_specific_setup(circuit, rng)?;
-        // save the proving key to a file in the output directory
-        tokio::fs::create_dir_all(&args.output_dir).await?;
-        let proving_key_path = args.output_dir.join("proving_key.bin");
-        let mut bytes = Vec::new();
-        pk.serialize_compressed(&mut bytes)?;
-        tokio::fs::write(proving_key_path, bytes).await?;
-    };
-
-    let circom = builder.build().unwrap();
-    let full_assignment = circom.witness.clone().unwrap();
-    let cs = ConstraintSystem::<Bn254Fr>::new_ref();
-    circom.generate_constraints(cs.clone()).unwrap();
-    assert!(cs.is_satisfied().unwrap());
-    let matrices = cs.to_matrices().unwrap();
-
-    let pp = PackedSharingParams::new(2);
-    let num_inputs = matrices.num_instance_variables;
-    let num_constraints = matrices.num_constraints;
-    let qap_timer = start_timer!(|| "QAP");
-    let qap = qap::<Bn254Fr, Radix2EvaluationDomain<_>>(&matrices, &full_assignment).unwrap();
-    end_timer!(qap_timer);
-    let qap_shares = qap.pss(&pp);
-    let ax_shares = pack_from_witness::<Bn254>(&pp, full_assignment[num_inputs..].to_vec());
-    let a_shares = pack_from_witness::<Bn254>(&pp, full_assignment[1..].to_vec());
-    let encoding_timer = start_timer!(|| "Encoding Shares");
-    // Prepare things to be written to disk
-    let qap_shares = qap_shares
-        .into_iter()
-        .map(|s| PackedQAPShare {
-            a: cfg_into_iter!(s.a)
-                .map(f_to_bytes)
-                .map(ByteBuf::from)
-                .collect(),
-            b: cfg_into_iter!(s.b)
-                .map(f_to_bytes)
-                .map(ByteBuf::from)
-                .collect(),
-            c: cfg_into_iter!(s.c)
-                .map(f_to_bytes)
-                .map(ByteBuf::from)
-                .collect(),
-        })
-        .collect::<Vec<_>>();
-    let ax_shares = cfg_into_iter!(ax_shares)
-        .map(|s| {
-            cfg_into_iter!(s)
-                .map(f_to_bytes)
-                .map(ByteBuf::from)
-                .collect::<Vec<_>>()
-        })
-        .collect::<Vec<_>>();
-    let a_shares = cfg_into_iter!(a_shares)
-        .map(|s| {
-            cfg_into_iter!(s)
-                .map(f_to_bytes)
-                .map(ByteBuf::from)
-                .collect::<Vec<_>>()
-        })
-        .collect::<Vec<_>>();
-    end_timer!(encoding_timer);
-
-    let circuit = CircuitProperties {
-        circuit_id: args.circuit_id,
-        pk_uri: tokio::fs::canonicalize(args.output_dir.join("proving_key.bin")).await?,
-        wasm_uri: tokio::fs::canonicalize(args.wasm).await?,
-        r1cs_uri: tokio::fs::canonicalize(args.r1cs).await?,
-        num_inputs,
-        num_constraints,
-    };
-    let job = JobProperties {
-        job_id: args.job_id,
-        circuit_id: args.circuit_id,
-        public_inputs,
-        pss_l: pp.l,
-        a_shares,
-        ax_shares,
-        qap_shares,
-    };
-
-    // write the circuit and job properties to a file
-    let job_path = args.output_dir.join(format!("job_{}.json", args.job_id));
-    let circuit_path = args
-        .output_dir
-        .join(format!("circuit_{}.json", args.circuit_id));
-    tokio::fs::write(job_path, serde_json::to_string(&job)?).await?;
-    tokio::fs::write(circuit_path, serde_json::to_string(&circuit)?).await?;
-    tracing::info!("Job and circuit properties written to disk");
-    end_timer!(setup_timer);
-    Ok(())
-}
-
-fn f_to_bytes<F: Field + CanonicalSerialize>(f: F) -> Vec<u8> {
-    let mut bytes = Vec::new();
-    f.serialize_compressed(&mut bytes).unwrap();
-    bytes
-}
-
-pub fn setup_log() {
-    let _ = SubscriberBuilder::default()
-        .with_env_filter(EnvFilter::from_default_env())
-        .finish()
-        .try_init();
-}
diff --git a/playground/examples/node.rs b/playground/examples/node.rs
deleted file mode 100644
index 4d6da2d46..000000000
--- a/playground/examples/node.rs
+++ /dev/null
@@ -1,544 +0,0 @@
-use std::{
-    collections::HashMap,
-    net::SocketAddr,
-    path::PathBuf,
-    str::FromStr,
-    sync::{
-        atomic::{AtomicU64, Ordering},
-        Arc,
-    },
-    time::Duration,
-};
-
-use ark_circom::CircomReduction;
-use ark_crypto_primitives::snark::SNARK;
-use ark_ec::pairing::Pairing;
-use ark_ec::CurveGroup;
-use ark_groth16::{Groth16, ProvingKey};
-use ark_poly::{EvaluationDomain, Radix2EvaluationDomain};
-use ark_relations::r1cs::SynthesisError;
-use ark_serialize::{CanonicalDeserialize, CanonicalSerialize};
-use ark_std::{cfg_iter, start_timer};
-use ark_std::{end_timer, Zero};
-use clap::Parser;
-use futures::{future, TryFutureExt};
-use gadget_core::gadget::substrate::Client;
-use gadget_core::job::{BuiltExecutableJobWrapper, JobBuilder, JobError};
-use groth16::proving_key::PackedProvingKeyShare;
-use mpc_net::prod::CertToDer;
-use mpc_net::MultiplexedStreamID;
-use sc_client_api::FinalizeSummary;
-use sc_utils::mpsc::{TracingUnboundedReceiver, TracingUnboundedSender};
-use secret_sharing::pss::PackedSharingParams;
-use sp_runtime::app_crypto::sp_core::Encode;
-use sp_runtime::codec::Decode;
-use sp_runtime::traits::Block;
-use tokio::sync::{mpsc::UnboundedSender, Mutex, Semaphore};
-use tracing_subscriber::{fmt::SubscriberBuilder, util::SubscriberInitExt, EnvFilter};
-use uuid::Uuid;
-use webb_gadget::{BlockImportNotification, FinalityNotification};
-use zk_gadget::{
-    client_ext::{
-        job_types::{CircuitProperties, JobProperties},
-        ClientWithApi,
-    },
-    network::ZkNetworkService,
-    protocol::proto_gen::ZkAsyncProtocolParameters,
-};
-
-#[cfg(feature = "parallel")]
-use rayon::prelude::*;
-
-const BLOCK_DURATION: Duration = Duration::from_millis(6000);
-
-type F = ark_bn254::Fr;
-type E = ark_bn254::Bn254;
-
-/// ZK Node
-#[derive(Parser, Debug)]
-#[command(author, version, about, long_about = None)]
-struct Args {
-    /// Name of the node, as it will appear in the logs
-    #[arg(long)]
-    name: String,
-    /// The Id of the node, also acts as the party index
-    #[arg(long)]
-    i: usize,
-    #[arg(long)]
-    n: usize,
-    /// King's IP address
-    #[arg(long)]
-    king_ip: SocketAddr,
-    /// Cuurent node's private identity DER
-    #[arg(long)]
-    private_identity_der: PathBuf,
-    /// All the nodes' public identity DER (including the current node)
-    /// it must be in the same order as the nodes' Ids
-    #[arg(long)]
-    public_identity_der: PathBuf,
-    #[arg(long)]
-    client_only_king_public_identity_der: Option<PathBuf>,
-    /// Watch directory for new jobs
-    #[arg(long)]
-    watch_dir: PathBuf,
-}
-
-#[tokio::main]
-async fn main() -> anyhow::Result<()> {
-    setup_log();
-    let args = Args::parse();
-    assert!(args.i < args.n, "i must be less than n");
-
-    tracing::info!(node = args.name, "Starting zknode");
-
-    let state = State {
-        circuits: Arc::new(Mutex::new(HashMap::new())),
-        jobs: Arc::new(Mutex::new(HashMap::new())),
-        processing_semophore: Arc::new(Semaphore::new(1)),
-        watch_dir: args.watch_dir,
-    };
-
-    let client = BlockchainClient::new(state.clone());
-    let (done_tx, mut done_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
-
-    let (public_identity_der, private_identity_der) = {
-        let public_identity_der_path = args.public_identity_der;
-        let private_identity_der_path = args.private_identity_der;
-        (
-            std::fs::read(public_identity_der_path)?,
-            std::fs::read(private_identity_der_path)?,
-        )
-    };
-
-    let gadget_config = if args.i == 0 {
-        zk_gadget::ZkGadgetConfig {
-            king_bind_addr: Some(args.king_ip),
-            client_only_king_addr: None,
-            id: args.i as _,
-            n_parties: args.n,
-            public_identity_der,
-            private_identity_der,
-            client_only_king_public_identity_der: None,
-        }
-    } else {
-        let king_public_identity_der_path = args
-            .client_only_king_public_identity_der
-            .expect("King's public identity DER path is required");
-        let king_public_identity_der = std::fs::read(king_public_identity_der_path)?;
-        zk_gadget::ZkGadgetConfig {
-            king_bind_addr: None,
-            client_only_king_addr: Some(args.king_ip),
-            id: args.i as _,
-            n_parties: args.n,
-            public_identity_der,
-            private_identity_der,
-            client_only_king_public_identity_der: Some(king_public_identity_der.clone()),
-        }
-    };
-
-    let additional_parameters = AdditionalParams {
-        stop_tx: done_tx.clone(),
-    };
-
-    let zk_gadget_future = zk_gadget::run(
-        gadget_config,
-        client,
-        additional_parameters,
-        async_protocol_generator,
-    );
-
-    let done_rx_future = async move {
-        done_rx.recv().await.ok_or("Did not receive done signal")?;
-        tracing::info!("Received done signal");
-        Ok::<_, String>(())
-    };
-
-    tokio::select! {
-        res0 = zk_gadget_future => {
-            res0?;
-        },
-        res1 = done_rx_future => {
-            res1.map_err(|e| anyhow::anyhow!("{e}"))?
-        }
-    }
-
-    Ok(())
-}
-
-#[derive(Clone)]
-struct State {
-    pub circuits: Arc<Mutex<HashMap<u64, CircuitProperties>>>,
-    pub jobs: Arc<Mutex<HashMap<[u8; 32], JobProperties>>>,
-    pub processing_semophore: Arc<Semaphore>,
-    pub watch_dir: PathBuf,
-}
-
-#[derive(Clone)]
-struct AdditionalParams {
-    #[allow(unused)]
-    stop_tx: UnboundedSender<()>,
-}
-
-fn async_protocol_generator(
-    params: ZkAsyncProtocolParameters<
-        AdditionalParams,
-        ZkNetworkService,
-        BlockchainClient,
-        TestBlock,
-    >,
-) -> BuiltExecutableJobWrapper {
-    let stop_tx = params.extra_parameters.stop_tx.clone();
-    let party_id = params.party_id;
-    JobBuilder::default()
-        .post(async move {
-            if party_id == 0 {
-                // If we are the king, wait a little bit for the other nodes to safely shutdown
-                tokio::time::sleep(Duration::from_millis(1000)).await;
-            }
-
-            stop_tx.send(()).expect("Failed to send stop signal");
-            Ok(())
-        })
-        .build(async move {
-        let _permit = match params.client.state.processing_semophore.try_acquire() {
-            Ok(permit) => permit,
-            Err(tokio::sync::TryAcquireError::NoPermits) => {
-                // No more processing power, we return.
-                return Ok(());
-            }
-            Err(tokio::sync::TryAcquireError::Closed) => unreachable!("Never close"),
-        };
-        let jobs = params.client.state.jobs.lock().await;
-        let circuits = params.client.state.circuits.lock().await;
-        let Some(job) = jobs.get(&params.associated_task_id).cloned() else {
-            return Err(JobError::from(format!(
-                "Job with id {} not found",
-                hex::encode(params.associated_task_id)
-            )));
-        };
-        let Some(circuit) = circuits.get(&job.circuit_id).cloned() else {
-            return Err(JobError::from(format!("Circuit with id {} not found", job.circuit_id)));
-        };
-
-        drop(jobs);
-        drop(circuits);
-
-        let proving_key = tokio::fs::read(&circuit.pk_uri)
-            .map_err(|e| JobError::from(format!("Failed to read proving key: {e}")))
-            .await?;
-        let pk = ProvingKey::deserialize_compressed(proving_key.as_slice()).map_err(|e| {
-            JobError::from(format!("Failed to deserialize proving key: {e}"))
-        })?;
-
-        let pp = PackedSharingParams::new(job.pss_l);
-        let pp_g1 = PackedSharingParams::new(pp.l);
-        let pp_g2 = PackedSharingParams::new(pp.l);
-        let crs_shares_timer = start_timer!(|| "CRS Shares");
-        let crs_shares =
-            PackedProvingKeyShare::<E>::pack_from_arkworks_proving_key(&pk, pp_g1, pp_g2);
-        end_timer!(crs_shares_timer);
-        let our_qap_share = job
-            .qap_shares
-            .get(params.party_id as usize)
-            .ok_or_else(|| JobError::from(format!("QAP Shares for party {} not found", params.party_id)))?;
-        let our_a_share = job.a_shares.get(params.party_id as usize).ok_or_else(|| JobError::from(format!("a Shares for party {} not found", params.party_id)))?;
-        let our_ax_share = job.ax_shares.get(params.party_id as usize).ok_or_else(|| JobError::from(format!("ax Shares for party {} not found", params.party_id)))?;
-        let m = circuit.num_inputs + circuit.num_constraints;
-        let domain = Radix2EvaluationDomain::new(m)
-            .ok_or(SynthesisError::PolynomialDegreeTooLarge)
-            .unwrap();
-        let qap_share = groth16::qap::PackedQAPShare {
-            num_inputs: circuit.num_inputs,
-            num_constraints: circuit.num_constraints,
-            a: cfg_iter!(our_qap_share.a)
-                .map(|f| F::deserialize_compressed(f.as_slice()))
-                .collect::<Result<Vec<_>, _>>()
-                .unwrap(),
-            b: cfg_iter!(our_qap_share.b)
-                .map(|f| F::deserialize_compressed(f.as_slice()))
-                .collect::<Result<Vec<_>, _>>()
-                .unwrap(),
-            c: cfg_iter!(our_qap_share.c)
-                .map(|f| F::deserialize_compressed(f.as_slice()))
-                .collect::<Result<Vec<_>, _>>()
-                .unwrap(),
-            domain,
-        };
-
-        let crs_share = crs_shares.get(params.party_id as usize).ok_or_else(|| JobError::from(format!("CRS Shares for party {} not found", params.party_id)))?;
-        let a_share = cfg_iter!(our_a_share)
-            .map(|f| F::deserialize_compressed(f.as_slice()))
-            .collect::<Result<Vec<_>, _>>()
-            .unwrap();
-        let ax_share = cfg_iter!(our_ax_share)
-            .map(|f| F::deserialize_compressed(f.as_slice()))
-            .collect::<Result<Vec<_>, _>>()
-            .unwrap();
-        let h_share = groth16::ext_wit::h(qap_share, &pp, &params).await.unwrap();
-        tracing::debug!("Finished Computing h_share");
-        let msm_section = start_timer!(|| "MSM operations");
-        // Compute msm while dropping the base vectors as they are not used again
-        let compute_a = start_timer!(|| "Compute A");
-        let pi_a_share = groth16::prove::A::<E> {
-            L: Default::default(),
-            N: Default::default(),
-            r: <E as Pairing>::ScalarField::zero(),
-            pp: &pp,
-            S: &crs_share.s,
-            a: &a_share,
-        }
-        .compute(&params, MultiplexedStreamID::Zero)
-        .await
-        .unwrap();
-        end_timer!(compute_a);
-
-        let compute_b = start_timer!(|| "Compute B");
-        let pi_b_share = groth16::prove::B::<E> {
-            Z: Default::default(),
-            K: Default::default(),
-            s: <E as Pairing>::ScalarField::zero(),
-            pp: &pp,
-            V: &crs_share.v,
-            a: &a_share,
-        }
-        .compute(&params, MultiplexedStreamID::Zero)
-        .await
-        .unwrap();
-        end_timer!(compute_b);
-
-        let compute_c = start_timer!(|| "Compute C");
-        let pi_c_share = groth16::prove::C::<E> {
-            W: &crs_share.w,
-            U: &crs_share.u,
-            A: pi_a_share,
-            M: Default::default(),
-            r: <E as Pairing>::ScalarField::zero(),
-            s: <E as Pairing>::ScalarField::zero(),
-            pp: &pp,
-            H: &crs_share.h,
-            a: &a_share,
-            ax: &ax_share,
-            h: &h_share,
-        }
-        .compute(&params)
-        .await
-        .unwrap();
-        end_timer!(compute_c);
-
-        end_timer!(msm_section);
-
-        // We only save the result if we are the king node
-        if params.party_id == 0 {
-            let (mut a, mut b, c) = (pi_a_share, pi_b_share, pi_c_share);
-            // These elements are needed to construct the full proof, they are part of the proving key.
-            // however, we can just send these values to the client, not the full proving key.
-            a += pk.a_query[0] + pk.vk.alpha_g1;
-            b += pk.b_g2_query[0] + pk.vk.beta_g2;
-
-            let proof = ark_groth16::Proof::<E> {
-                a: a.into_affine(),
-                b: b.into_affine(),
-                c: c.into_affine(),
-            };
-
-            tracing::info!(job_id = %job.job_id, a = %proof.a, b = %proof.b, c = %proof.c, "Proof Generated");
-            // Verify the proof
-            // convert the public inputs from string to bigints
-            let public_inputs = job
-                .public_inputs
-                .iter()
-                .map(|s| F::from_str(s).unwrap())
-                .collect::<Vec<_>>();
-            let pvk = ark_groth16::prepare_verifying_key(&pk.vk);
-            let verified = Groth16::<E, CircomReduction>::verify_with_processed_vk(
-                &pvk,
-                public_inputs.as_slice(),
-                &proof,
-            )
-            .unwrap();
-            if verified {
-                tracing::info!("Proof verified");
-            } else {
-                tracing::error!("Proof verification failed");
-            }
-            let mut proof_bytes = Vec::new();
-            proof.serialize_compressed(&mut proof_bytes).unwrap();
-            let proof_path = params
-                .client
-                .state
-                .watch_dir
-                .join(format!("proof_{}", job.job_id));
-            tokio::fs::write(proof_path, proof_bytes).await.unwrap();
-            tracing::info!("Proof written to file");
-        }
-        // TODO: remove the job from the queue/disk
-        Ok(())
-    })
-}
-
-#[derive(Clone)]
-struct BlockchainClient {
-    latest_received_header: Arc<AtomicU64>,
-    state: State,
-    tx: TracingUnboundedSender<<TestBlock as Block>::Hash>,
-    #[allow(unused)]
-    rx: Arc<TracingUnboundedReceiver<<TestBlock as Block>::Hash>>,
-}
-
-impl BlockchainClient {
-    fn new(state: State) -> Self {
-        let (tx, rx) = sc_utils::mpsc::tracing_unbounded::<<TestBlock as Block>::Hash>(
-            "mpsc_finality_notification",
-            999999,
-        );
-        Self {
-            latest_received_header: Arc::new(AtomicU64::new(0)),
-            state,
-            tx,
-            rx: Arc::new(rx),
-        }
-    }
-}
-
-#[async_trait::async_trait]
-impl Client<TestBlock> for BlockchainClient {
-    async fn get_next_finality_notification(&self) -> Option<FinalityNotification<TestBlock>> {
-        self.get_latest_finality_notification().await
-    }
-
-    async fn get_latest_finality_notification(&self) -> Option<FinalityNotification<TestBlock>> {
-        let _permit = self.state.processing_semophore.acquire().await.unwrap();
-
-        let mut jobs = self.state.jobs.lock().await;
-        let mut circuits = self.state.circuits.lock().await;
-        let mut entries = tokio::fs::read_dir(&self.state.watch_dir).await.unwrap();
-
-        // Fetch any new jobs from the watch directory.
-        // by walking the directory and reading the job files and circuit files
-        // and adding them to the state
-        while let Ok(Some(entry)) = entries.next_entry().await {
-            let path = entry.path();
-            let file_name = path.file_name().unwrap().to_str().unwrap();
-            // job_{id}.json or circuit_{id}.json
-            let parts = file_name.split('_').collect::<Vec<_>>();
-            // job or circuit
-            let prefix = parts[0];
-            // if the file is not a job or circuit file, skip it
-            if prefix != "job" && prefix != "circuit" {
-                continue;
-            }
-            let id = parts[1].split('.').collect::<Vec<_>>()[0]
-                .parse::<u64>()
-                .unwrap();
-            // convert the id to 32 bytes hash.
-            let mut id_bytes = [0u8; 32];
-            id_bytes[..8].copy_from_slice(&id.to_be_bytes());
-            match prefix {
-                "job" if jobs.contains_key(&id_bytes) => {
-                    tracing::trace!("Job {} already exists", id);
-                }
-                "job" => {
-                    // parse the job file
-                    let job = tokio::fs::read_to_string(path).await.unwrap();
-                    let job: JobProperties = serde_json::from_str(&job).unwrap();
-                    jobs.insert(id_bytes, job);
-                    tracing::debug!("Added job {} to state", id);
-                }
-                "circuit" if circuits.contains_key(&id) => {
-                    tracing::trace!("Circuit {} already exists", id);
-                }
-                "circuit" => {
-                    let circuit = tokio::fs::read_to_string(path).await.unwrap();
-                    let circuit: CircuitProperties = serde_json::from_str(&circuit).unwrap();
-                    circuits.insert(circuit.circuit_id, circuit);
-                    tracing::debug!("Added circuit {} to state", id);
-                }
-                _ => {}
-            }
-        }
-        let v = self.latest_received_header.fetch_add(1, Ordering::AcqRel);
-        // Sleep for a while to simulate block time
-        tokio::time::sleep(BLOCK_DURATION).await;
-        Some(block_number_to_finality_notification(v, self.tx.clone()))
-    }
-
-    async fn get_next_block_import_notification(
-        &self,
-    ) -> Option<BlockImportNotification<TestBlock>> {
-        future::pending().await
-    }
-}
-
-#[async_trait::async_trait]
-impl ClientWithApi<TestBlock> for BlockchainClient {
-    async fn get_job_circuit_properties(
-        &self,
-        _circuit_id: u64,
-    ) -> Result<Option<CircuitProperties>, webb_gadget::Error> {
-        Ok(None)
-    }
-
-    async fn get_job_properties(
-        &self,
-        _job_id: u64,
-    ) -> Result<Option<JobProperties>, webb_gadget::Error> {
-        Ok(None)
-    }
-
-    async fn get_next_job(&self) -> Result<Option<JobProperties>, webb_gadget::Error> {
-        let lock = self.state.jobs.lock().await;
-        let maybe_job = lock.values().next().cloned();
-        Ok(maybe_job)
-    }
-}
-
-pub type TestBlock = sp_runtime::testing::Block<XtDummy>;
-#[derive(Encode, Decode, sp_runtime::Serialize, Clone, Eq, PartialEq, Debug)]
-pub struct XtDummy;
-
-impl sp_runtime::traits::Extrinsic for XtDummy {
-    type Call = ();
-    type SignaturePayload = ();
-}
-
-fn block_number_to_finality_notification(
-    block_number: u64,
-    tx: TracingUnboundedSender<<TestBlock as Block>::Hash>,
-) -> FinalityNotification<TestBlock> {
-    let header = sp_runtime::generic::Header::<u64, _>::new_from_number(block_number);
-    let mut slice = [0u8; 32];
-    slice[..8].copy_from_slice(&block_number.to_be_bytes());
-    // add random uuid to ensure uniqueness
-    slice[8..24].copy_from_slice(&Uuid::new_v4().to_u128_le().to_be_bytes());
-
-    let hash = sp_runtime::testing::H256::from(slice);
-    let summary = FinalizeSummary {
-        header,
-        finalized: vec![hash],
-        stale_heads: vec![],
-    };
-
-    FinalityNotification::<TestBlock>::from_summary(summary, tx)
-}
-
-pub fn setup_log() {
-    let _ = SubscriberBuilder::default()
-        .with_env_filter(EnvFilter::from_default_env())
-        .finish()
-        .try_init();
-}
-
-struct CertBytes<'a> {
-    cert: &'a [u8],
-    private_key: &'a [u8],
-}
-
-impl<'a> CertToDer for CertBytes<'a> {
-    fn serialize_certificate_to_der(&self) -> Result<Vec<u8>, mpc_net::MpcNetError> {
-        Ok(self.cert.to_vec())
-    }
-
-    fn serialize_private_key_to_der(&self) -> Result<Vec<u8>, mpc_net::MpcNetError> {
-        Ok(self.private_key.to_vec())
-    }
-}
diff --git a/playground/generate_certs.sh b/playground/generate_certs.sh
deleted file mode 100755
index a47f9d4a4..000000000
--- a/playground/generate_certs.sh
+++ /dev/null
@@ -1,9 +0,0 @@
-#!/bin/sh
-# A simple script that invokes a simple binary to generate a self-signed
-# certificate for use in the playground.
-n=8
-for i in $(seq 0 $(($n - 1)))
-  do
-      cargo run --release --example certgen -- --i $i
-  done
-echo "Done generating certs"
diff --git a/playground/run.sh b/playground/run.sh
deleted file mode 100755
index da6466ead..000000000
--- a/playground/run.sh
+++ /dev/null
@@ -1,52 +0,0 @@
-#!/bin/sh
-set -e
-trap "exit" INT TERM
-trap "kill 0" EXIT
-
-# Step 0: cd to git working dir
-cd "$(git rev-parse --show-toplevel)"
-
-# Step 1: check to see if arg "--create-certs" was passed
-if [ "$1" = "--create-certs" ]; then
-    # Step 2: if so, invoke the generate_certs.sh script
-    echo "Creating certs..."
-    ./playground/generate_certs.sh
-fi
-
-# Step 2: Run the client unless --no-client was passed as the second or first parameter
-if [ "$1" != "--no-client" ] && [ "$2" != "--no-client" ]; then
-    cargo run --release --example zk-test-client -- --circuit-id 1 --job-id 1 --wasm ./fixtures/sha256/circom.wasm --r1cs ./fixtures/sha256/circom.r1cs --input ./fixtures/sha256/input.json --public-inputs ./fixtures/sha256/public_inputs.json --output-dir ./target/zk --generate-proving-key
-fi
-
-# Step 3: for x in 0..=7, run a command
-PROCS=()
-SERVER_IP=127.0.0.1
-SERVER_PORT=5555
-n=8
-for i in $(seq 0 $(($n - 1)))
-  do
-    # if i is zero, run the king
-    if [ "$i" -eq "0" ]; then
-        # Step 4: Run the king
-        RUST_LOG=debug cargo run --release --example zknode -- --n $n --king-ip $SERVER_IP:$SERVER_PORT --public-identity-der certs/$i/cert.der --watch-dir target/zk --name king --i $i --private-identity-der certs/0/key.der &
-        pid=$!
-        PROCS[$i]=$pid
-
-        # But first, wait for the king to start
-        while ! nc -z -w 5 $SERVER_IP $SERVER_PORT; do
-          echo "Waiting for server $SERVER_IP:$SERVER_PORT ... "
-          sleep 0.5
-        done
-    else
-        # Step 4: Run the client
-        RUST_LOG=debug cargo run --release --example zknode -- --n $n --king-ip $SERVER_IP:$SERVER_PORT --public-identity-der certs/$i/cert.der --client-only-king-public-identity-der certs/0/cert.der --watch-dir target/zk --name node$i --i $i --private-identity-der certs/$i/key.der &
-        pid=$!
-        PROCS[$i]=$pid
-        sleep 0.2
-    fi
-  done
-
-# Step 5: wait for all processes to finish
-for pid in ${PROCS[@]}; do
-    wait $pid
-done
\ No newline at end of file
diff --git a/protocols/mp-ecdsa/Cargo.toml b/protocols/mp-ecdsa/Cargo.toml
new file mode 100644
index 000000000..6fdfe351d
--- /dev/null
+++ b/protocols/mp-ecdsa/Cargo.toml
@@ -0,0 +1,44 @@
+[package]
+name = "mp-ecdsa-protocol"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[features]
+application = []
+
+[dependencies]
+tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
+gadget-common = { workspace = true }
+gadget-core = { workspace = true }
+async-trait = { workspace = true }
+log = { workspace = true }
+parking_lot = { workspace = true }
+multi-party-ecdsa = { workspace = true }
+round-based = { workspace = true }
+curv = { workspace = true }
+futures = { workspace = true }
+itertools = { workspace = true }
+bincode2 = { workspace = true }
+linked-hash-map = { workspace = true }
+
+pallet-jobs-rpc-runtime-api = { workspace = true, features = ["std"] }
+tangle-primitives = { workspace = true, features = ["std"] }
+
+sp-core = { workspace = true, features = ["std"] }
+sp-api = { workspace = true, features = ["std"] }
+sp-runtime = { workspace = true, features = ["std"] }
+sp-application-crypto = { workspace = true, features = ["std"] }
+sp-consensus-aura = { workspace = true, features = ["std"] }
+
+sc-client-api = { workspace = true }
+sc-network = { workspace = true }
+sc-network-common = { workspace = true }
+sc-network-sync = { workspace = true }
+
+parity-scale-codec = { workspace = true }
+serde = { version = "1.0.193", features = ["derive"] }
+
+[build-dependencies]
+substrate-build-script-utils = { workspace = true }
\ No newline at end of file
diff --git a/protocols/mp-ecdsa/build.rs b/protocols/mp-ecdsa/build.rs
new file mode 100644
index 000000000..ea27b6219
--- /dev/null
+++ b/protocols/mp-ecdsa/build.rs
@@ -0,0 +1,21 @@
+// Copyright 2022 Webb Technologies Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+use substrate_build_script_utils::{generate_cargo_keys, rerun_if_git_head_changed};
+
+fn main() {
+    generate_cargo_keys();
+
+    rerun_if_git_head_changed();
+}
diff --git a/protocols/mp-ecdsa/src/constants.rs b/protocols/mp-ecdsa/src/constants.rs
new file mode 100644
index 000000000..9839fcf5c
--- /dev/null
+++ b/protocols/mp-ecdsa/src/constants.rs
@@ -0,0 +1,45 @@
+// Constants for dkg-gadget
+
+// ================= Common ======================== //
+pub const DKG_KEYGEN_PROTOCOL_NAME: &str = "/webb-tools/dkg/keygen/1";
+pub const DKG_SIGNING_PROTOCOL_NAME: &str = "/webb-tools/dkg/signing/1";
+
+// ============= Signing Worker ======================= //
+
+pub mod signing_worker {
+    use std::time::Duration;
+
+    // the maximum number of tasks that the work manager tries to assign
+    pub const MAX_RUNNING_TASKS: usize = 1;
+
+    // the maximum number of tasks that can be enqueued,
+    // enqueued here implies not actively running but listening for messages
+    pub const MAX_ENQUEUED_TASKS: usize = 20;
+
+    // How often to poll the jobs to check completion status
+    pub const JOB_POLL_INTERVAL: Duration = Duration::from_millis(500);
+}
+
+// ============= Networking ======================= //
+
+pub mod network {
+    /// Maximum number of known messages hashes to keep for a peer.
+    pub const MAX_KNOWN_MESSAGES: usize = 4096;
+
+    /// Maximum allowed size for a DKG Signed Message notification.
+    pub const MAX_MESSAGE_SIZE: u64 = 16 * 1024 * 1024;
+
+    /// Maximum number of duplicate messages that a single peer can send us.
+    ///
+    /// This is to prevent a malicious peer from spamming us with messages.
+    pub const MAX_DUPLICATED_MESSAGES_PER_PEER: usize = 8;
+}
+
+// ============= Keygen Worker ======================= //
+
+pub mod keygen_worker {
+    /// only 1 task at a time may run for keygen
+    pub const MAX_RUNNING_TASKS: usize = 1;
+    /// There should never be any job enqueueing for keygen
+    pub const MAX_ENQUEUED_TASKS: usize = 0;
+}
diff --git a/protocols/mp-ecdsa/src/error.rs b/protocols/mp-ecdsa/src/error.rs
new file mode 100644
index 000000000..3b2b51501
--- /dev/null
+++ b/protocols/mp-ecdsa/src/error.rs
@@ -0,0 +1,33 @@
+use gadget_core::job::JobError;
+use std::fmt::{Display, Formatter};
+
+#[derive(Debug)]
+pub enum Error {
+    Keystore(String),
+    Signature(String),
+    Job(JobError),
+    InvalidKeygenPartyId,
+    InvalidSigningSet,
+}
+
+impl Display for Error {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        std::fmt::Debug::fmt(self, f)
+    }
+}
+
+impl std::error::Error for Error {}
+
+impl From<JobError> for Error {
+    fn from(value: JobError) -> Self {
+        Error::Job(value)
+    }
+}
+
+impl From<Error> for JobError {
+    fn from(val: Error) -> Self {
+        JobError {
+            reason: val.to_string(),
+        }
+    }
+}
diff --git a/protocols/mp-ecdsa/src/lib.rs b/protocols/mp-ecdsa/src/lib.rs
new file mode 100644
index 000000000..2711b105f
--- /dev/null
+++ b/protocols/mp-ecdsa/src/lib.rs
@@ -0,0 +1,104 @@
+use gadget_common::client::{AccountId, ClientWithApi};
+use gadget_common::debug_logger::DebugLogger;
+use gadget_common::gadget::network::Network;
+use gadget_common::keystore::{ECDSAKeyStore, KeystoreBackend};
+use pallet_jobs_rpc_runtime_api::JobsApi;
+use sc_client_api::Backend;
+use sp_api::ProvideRuntimeApi;
+use sp_runtime::traits::Block;
+use std::error::Error;
+use std::sync::Arc;
+
+pub mod constants;
+pub mod error;
+pub mod network;
+pub mod protocols;
+pub mod util;
+
+pub struct MpEcdsaProtocolConfig {
+    pub account_id: AccountId,
+    pub id: sp_core::ecdsa::Public,
+}
+
+pub async fn run_keygen<B, BE, KBE, C, N>(
+    config: &MpEcdsaProtocolConfig,
+    client_inner: Arc<C>,
+    logger: DebugLogger,
+    keystore: ECDSAKeyStore<KBE>,
+    network: N,
+) -> Result<(), Box<dyn Error>>
+where
+    B: Block,
+    BE: Backend<B> + 'static,
+    C: ClientWithApi<B, BE>,
+    KBE: KeystoreBackend,
+    N: Network,
+    <C as ProvideRuntimeApi<B>>::Api: JobsApi<B, AccountId>,
+{
+    let client =
+        gadget_common::client::create_client(client_inner.clone(), logger.clone(), keystore)
+            .await?;
+    let protocol =
+        protocols::keygen::create_protocol(config, client, network.clone(), logger).await?;
+
+    gadget_common::run_protocol(network, protocol, client_inner).await?;
+    Ok(())
+}
+
+pub async fn run_sign<B, BE, KBE, C, N>(
+    config: &MpEcdsaProtocolConfig,
+    client_inner: Arc<C>,
+    logger: DebugLogger,
+    keystore: ECDSAKeyStore<KBE>,
+    network: N,
+) -> Result<(), Box<dyn Error>>
+where
+    B: Block,
+    BE: Backend<B> + 'static,
+    C: ClientWithApi<B, BE>,
+    KBE: KeystoreBackend,
+    N: Network,
+    <C as ProvideRuntimeApi<B>>::Api: JobsApi<B, AccountId>,
+{
+    let client =
+        gadget_common::client::create_client(client_inner.clone(), logger.clone(), keystore)
+            .await?;
+    let protocol =
+        protocols::sign::create_protocol(config, logger, client, network.clone()).await?;
+
+    gadget_common::run_protocol(network, protocol, client_inner).await?;
+    Ok(())
+}
+
+pub async fn run<B, BE, KBE, C, N, N2>(
+    config: MpEcdsaProtocolConfig,
+    client: C,
+    logger: DebugLogger,
+    keystore: ECDSAKeyStore<KBE>,
+    network_keygen: N,
+    network_signing: N2,
+) -> Result<(), Box<dyn Error>>
+where
+    B: Block,
+    BE: Backend<B> + 'static,
+    C: ClientWithApi<B, BE>,
+    KBE: KeystoreBackend,
+    N: Network,
+    N2: Network,
+    <C as ProvideRuntimeApi<B>>::Api: JobsApi<B, AccountId>,
+{
+    let client = Arc::new(client);
+    let keygen_future = run_keygen(
+        &config,
+        client.clone(),
+        logger.clone(),
+        keystore.clone(),
+        network_keygen,
+    );
+    let sign_future = run_sign(&config, client, logger, keystore, network_signing);
+
+    tokio::select! {
+        res0 = keygen_future => res0,
+        res1 = sign_future => res1,
+    }
+}
diff --git a/protocols/mp-ecdsa/src/network.rs b/protocols/mp-ecdsa/src/network.rs
new file mode 100644
index 000000000..a7f69e3ac
--- /dev/null
+++ b/protocols/mp-ecdsa/src/network.rs
@@ -0,0 +1,802 @@
+// Copyright 2022 Webb Technologies Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! A DKG Gossip Engine that uses [`sc_network::NetworkService`] as a backend.
+//!
+//! In a nutshell, it works as follows:
+//!
+//! 1. You create a new [`NetworkGossipEngineBuilder`] which does not require any setup for now,
+//! 2. You call [`NetworkGossipEngineBuilder::build`] to get two things:
+//!   - a [`GossipHandler`] that is a simple background task that should run indefinitely, and
+//!   - a [`GossipHandlerController`] that can be used to control that background task.
+//!
+//! The background task ([`GossipHandler`]) role is to listen, and gossip (if enabled) all the
+//! DKG messages.
+//!
+//! From the [`GossipHandlerController`] which implements [`super::GossipEngineIface`], you can:
+//!  - send a DKG message to a specific peer.
+//!  - send a DKG message to all peers.
+//!  - get a notification stream when you get a DKG message.
+//!  - Have access to the message queue, which is a FIFO queue of DKG messages.
+//!
+//!
+//! ### The Lifetime of the DKG Message:
+//!
+//! The DKG message is a [`SignedDKGMessage`] that is signed by the DKG authority, first it get
+//! sent to the Gossip Engine either by calling [`GossipHandlerController::send`] or
+//! [`GossipHandlerController::gossip`], depending on the call, the message will be sent to all
+//! peers or only to a specific peer. on the other end, the DKG message is received by the DKG
+//! engine, and it is verified then it will be added to the Engine's internal stream of DKG
+//! messages, later the DKG Gadget will read this stream and process the DKG message.
+
+pub use crate::constants::network::*;
+use async_trait::async_trait;
+use futures::StreamExt;
+use gadget_common::client::AccountId;
+use gadget_common::debug_logger::DebugLogger;
+use gadget_common::gadget::message::GadgetProtocolMessage;
+use gadget_common::gadget::network::Network;
+use gadget_common::gadget::work_manager::WebbWorkManager;
+use gadget_common::keystore::{ECDSAKeyStore, KeystoreBackend};
+use gadget_common::Error;
+use gadget_core::job_manager::WorkManagerInterface;
+use linked_hash_map::LinkedHashMap;
+use parking_lot::{Mutex, RwLock};
+use sc_network::{
+    config, error, multiaddr, Event, NetworkEventStream, NetworkNotification, NetworkPeers,
+    NetworkService, NetworkStateInfo, PeerId, ProtocolName, SyncEventStream,
+};
+use sc_network_common::sync::{Metrics, SyncEvent};
+use sc_network_sync::SyncingService;
+use serde::{Deserialize, Serialize};
+use sp_runtime::traits::Block;
+use std::{
+    collections::{HashMap, HashSet},
+    hash::Hash,
+    iter,
+    num::NonZeroUsize,
+    sync::{
+        atomic::{AtomicBool, Ordering},
+        Arc,
+    },
+};
+use tokio::sync::mpsc::UnboundedReceiver;
+
+#[derive(Clone)]
+#[allow(dead_code)]
+pub struct NetworkGossipEngineBuilder<KBE: KeystoreBackend> {
+    protocol_name: ProtocolName,
+    keystore: ECDSAKeyStore<KBE>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum GossipNetworkMessage {
+    Handshake(HandshakeMessage),
+    Protocol(GadgetProtocolMessage),
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct HandshakeMessage {
+    pub account_id: AccountId,
+}
+
+impl<KBE: KeystoreBackend> NetworkGossipEngineBuilder<KBE> {
+    /// Create a new network gossip engine.
+    pub fn new(protocol_name: ProtocolName, keystore: ECDSAKeyStore<KBE>) -> Self {
+        Self {
+            protocol_name,
+            keystore,
+        }
+    }
+
+    /// Returns the configuration of the set to put in the network configuration.
+    pub fn set_config(protocol_name: ProtocolName) -> config::NonDefaultSetConfig {
+        config::NonDefaultSetConfig {
+            handshake: None,
+            notifications_protocol: protocol_name,
+            fallback_names: Vec::new(),
+            max_notification_size: MAX_MESSAGE_SIZE,
+            set_config: config::SetConfig {
+                in_peers: 0,
+                out_peers: 0,
+                reserved_nodes: Vec::new(),
+                non_reserved_mode: config::NonReservedPeerMode::Deny,
+            },
+        }
+    }
+
+    /// Turns the builder into the actual handler. Returns a controller that allows controlling
+    /// the behaviour of the handler while it's running.
+    ///
+    /// Important: the gossip mechanism is initially disabled and doesn't gossip messages.
+    /// You must call [`GossipHandlerController::set_gossip_enabled`] to enable it.
+    ///
+    /// The returned values are:
+    /// - a [`GossipHandler`] that is a simple background task that should run indefinitely, and
+    /// - a [`GossipHandlerController`] that can be used to control that background task.
+    pub fn build<B: Block>(
+        self,
+        service: Arc<NetworkService<B, B::Hash>>,
+        sync_service: Arc<SyncingService<B>>,
+        metrics: Option<Metrics>,
+        logger: DebugLogger,
+        account_id: AccountId,
+    ) -> error::Result<(GossipHandler<B, KBE>, GossipHandlerController)> {
+        // Here we need to create few channels to communicate back and forth between the
+        // background task and the controller.
+        // since we have two things here we will need two channels:
+        // 1. a channel to send commands to the background task (Controller -> Background).
+        let (handler_channel, handler_channel_rx) = tokio::sync::mpsc::unbounded_channel();
+        let (message_channel_tx, message_channel_rx) = tokio::sync::mpsc::unbounded_channel();
+        let gossip_enabled = Arc::new(AtomicBool::new(false));
+        let authority_id_to_peer_id = Arc::new(RwLock::new(HashMap::new()));
+        let handler = GossipHandler {
+            account_id,
+            keystore: self.keystore,
+            protocol_name: self.protocol_name.clone(),
+            to_receiver: message_channel_tx,
+            incoming_messages_stream: Arc::new(Mutex::new(Some(handler_channel_rx))),
+            pending_messages_peers: Arc::new(RwLock::new(LruHashMap::new(
+                NonZeroUsize::new(MAX_KNOWN_MESSAGES).expect("Constant is nonzero"),
+            ))),
+            authority_id_to_peer_id: authority_id_to_peer_id.clone(),
+            gossip_enabled: gossip_enabled.clone(),
+            service,
+            sync_service,
+            peers: Arc::new(RwLock::new(HashMap::new())),
+            logger: logger.clone(),
+            metrics: Arc::new(metrics),
+        };
+        let local_peer_id = handler.service.local_peer_id();
+
+        let controller = GossipHandlerController {
+            local_peer_id,
+            authority_id_to_peer_id,
+            protocol_name: self.protocol_name,
+            handler_channel,
+            message_notifications_channel: Arc::new(tokio::sync::Mutex::new(Some(
+                message_channel_rx,
+            ))),
+            gossip_enabled,
+            logger,
+        };
+
+        Ok((handler, controller))
+    }
+}
+
+#[allow(unused)]
+mod rep {
+    use sc_network::ReputationChange as Rep;
+    /// Reputation change when a peer sends us a message that we didn't know about.
+    pub const GOOD_MESSAGE: Rep = Rep::new(1 << 7, "Good message");
+    /// We received an unexpected message packet.
+    pub const UNEXPECTED_MESSAGE: Rep = Rep::new_fatal("Unexpected message packet");
+    /// When a peer tries to impersonate another peer, by claiming another authority id.
+    pub const PEER_IMPERSONATED: Rep = Rep::new_fatal("Peer is impersonating another peer");
+    /// Reputation change when a peer sends us the same message over and over.
+    pub const DUPLICATE_MESSAGE: Rep = Rep::new(-(1 << 12), "Duplicate message");
+}
+
+/// Controls the behaviour of a [`GossipHandler`] it is connected to.
+#[derive(Clone)]
+#[allow(dead_code)]
+pub struct GossipHandlerController {
+    local_peer_id: PeerId,
+    protocol_name: ProtocolName,
+    /// a channel to send commands to the background task (Controller -> Background).
+    handler_channel: tokio::sync::mpsc::UnboundedSender<ToHandler>,
+    /// where messages are received
+    message_notifications_channel:
+        Arc<tokio::sync::Mutex<Option<UnboundedReceiver<GossipNetworkMessage>>>>,
+    /// Whether the gossip mechanism is enabled or not.
+    gossip_enabled: Arc<AtomicBool>,
+    authority_id_to_peer_id: Arc<RwLock<HashMap<AccountId, PeerId>>>,
+    logger: DebugLogger,
+}
+
+#[async_trait]
+impl Network for GossipHandlerController {
+    async fn next_message(
+        &self,
+    ) -> Option<<WebbWorkManager as WorkManagerInterface>::ProtocolMessage> {
+        loop {
+            let message = self
+                .message_notifications_channel
+                .lock()
+                .await
+                .as_mut()?
+                .recv()
+                .await?;
+
+            match message {
+                GossipNetworkMessage::Handshake(..) => {
+                    self.logger
+                        .warn("Should not receive a handshake message here");
+                }
+                GossipNetworkMessage::Protocol(protocol_message) => return Some(protocol_message),
+            }
+        }
+    }
+
+    async fn send_message(
+        &self,
+        message: <WebbWorkManager as WorkManagerInterface>::ProtocolMessage,
+    ) -> Result<(), Error> {
+        // Make sure that we programmed-in the account IDs
+        let _ = message.from_account_id.ok_or_else(|| Error::NetworkError {
+            err: "Improperly constructed message due to no associated account IDs".to_string(),
+        })?;
+
+        if let Some(to) = message.to_account_id.as_ref() {
+            let id = self
+                .authority_id_to_peer_id
+                .read()
+                .get(to)
+                .cloned()
+                .ok_or_else(|| Error::NetworkError {
+                    err: format!("Failed to send message to peer due to no connection: {to:?}"),
+                })?;
+
+            self.handler_channel
+                .send(ToHandler::SendMessage {
+                    recipient: id,
+                    message: GossipNetworkMessage::Protocol(message),
+                })
+                .map_err(|err| Error::NetworkError {
+                    err: format!("Failed to send message to handler: {err:?}"),
+                })
+        } else {
+            self.handler_channel
+                .send(ToHandler::Gossip(GossipNetworkMessage::Protocol(message)))
+                .map_err(|err| Error::NetworkError {
+                    err: format!("Failed to send message to handler: {err:?}"),
+                })
+        }
+    }
+
+    async fn run(&self) -> Result<(), Error> {
+        futures::future::pending().await
+    }
+}
+/// an Enum Representing the commands that can be sent to the background task.
+#[derive(Clone, Debug)]
+enum ToHandler {
+    /// Send a DKG message to a peer.
+    SendMessage {
+        recipient: PeerId,
+        message: GossipNetworkMessage,
+    },
+    /// Gossip a DKG message to all peers.
+    Gossip(GossipNetworkMessage),
+}
+
+impl GossipHandlerController {
+    /// Controls whether messages are being gossiped on the network.
+    pub fn set_gossip_enabled(&self, enabled: bool) {
+        self.gossip_enabled.store(enabled, Ordering::Relaxed);
+    }
+}
+
+/// Handler for gossiping messages. Call [`GossipHandler::run`] to start the processing.
+///
+/// This is a background task that handles all the DKG messages.
+pub struct GossipHandler<B: Block + 'static, KBE: KeystoreBackend> {
+    /// The Protocol Name, should be unique.
+    ///
+    /// Used as an identifier for the gossip protocol.
+    protocol_name: ProtocolName,
+    /// The DKG Keystore.
+    keystore: ECDSAKeyStore<KBE>,
+    /// A Simple notification stream to notify the caller that we have messages in the queue.
+    to_receiver: tokio::sync::mpsc::UnboundedSender<GossipNetworkMessage>,
+    incoming_messages_stream: Arc<Mutex<Option<UnboundedReceiver<ToHandler>>>>,
+    /// As multiple peers can send us the same message, we group
+    /// these peers using the message hash while the message is
+    /// received. This prevents that we receive the same message
+    /// multiple times concurrently.
+    pending_messages_peers: Arc<RwLock<LruHashMap<Vec<u8>, HashSet<PeerId>>>>,
+    /// Network service to use to send messages and manage peers.
+    service: Arc<NetworkService<B, B::Hash>>,
+    sync_service: Arc<SyncingService<B>>,
+    // All connected peers
+    peers: Arc<RwLock<HashMap<PeerId, Peer>>>,
+    /// A mapping from authority id to peer id.
+    ///
+    /// This is used to send messages to specific peer by knowing the authority id.
+    authority_id_to_peer_id: Arc<RwLock<HashMap<AccountId, PeerId>>>,
+    /// Whether the gossip mechanism is enabled or not.
+    gossip_enabled: Arc<AtomicBool>,
+    logger: DebugLogger,
+    /// Prometheus metrics.
+    metrics: Arc<Option<Metrics>>,
+    account_id: AccountId,
+}
+
+impl<B: Block + 'static, KBE: KeystoreBackend> Clone for GossipHandler<B, KBE> {
+    fn clone(&self) -> Self {
+        Self {
+            account_id: self.account_id,
+            protocol_name: self.protocol_name.clone(),
+            keystore: self.keystore.clone(),
+            to_receiver: self.to_receiver.clone(),
+            incoming_messages_stream: self.incoming_messages_stream.clone(),
+            pending_messages_peers: self.pending_messages_peers.clone(),
+            service: self.service.clone(),
+            sync_service: self.sync_service.clone(),
+            peers: self.peers.clone(),
+            authority_id_to_peer_id: self.authority_id_to_peer_id.clone(),
+            gossip_enabled: self.gossip_enabled.clone(),
+            logger: self.logger.clone(),
+            metrics: self.metrics.clone(),
+        }
+    }
+}
+
+/// Peer information
+#[derive(Debug)]
+struct Peer {
+    /// Holds a set of messages known to this peer.
+    known_messages: LruHashSet<Vec<u8>>,
+    /// a counter of the messages that are received from this peer.
+    ///
+    /// Implemented as a HashMap/LruHashMap with the message hash as the key,
+    /// This is used to track the frequency of the messages received from this peer.
+    /// If the same message is received from this peer more than
+    /// `MAX_DUPLICATED_MESSAGES_PER_PEER`, we will flag this peer as malicious.
+    message_counter: LruHashMap<Vec<u8>, usize>,
+
+    /// The Known AuthorityId of that Peer.
+    ///
+    /// Could be None if that peer did not handshake with us yet.
+    authority_id: Option<AccountId>,
+}
+
+impl<B: Block + 'static, KBE: KeystoreBackend> GossipHandler<B, KBE> {
+    /// Turns the [`GossipHandler`] into a future that should run forever and not be
+    /// interrupted.
+    pub async fn run(self) {
+        let mut incoming_messages = self
+            .incoming_messages_stream
+            .lock()
+            .take()
+            .expect("incoming message stream already taken");
+        let mut event_stream = self.service.event_stream("dkg-handler");
+        let mut sync_event_stream = self.sync_service.event_stream("dkg-handler");
+
+        self.logger.debug("Starting the DKG Gossip Handler");
+
+        // we have two streams, one from the network and one from the controller.
+        // hence we want to start two separate tasks, one for each stream that are running
+        // in parallel, without blocking each other.
+
+        // first task, handles the incoming messages/Commands from the controller.
+        let self0 = self.clone();
+        let incoming_messages_task = tokio::spawn(async move {
+            while let Some(message) = incoming_messages.recv().await {
+                match message {
+                    ToHandler::SendMessage { recipient, message } => {
+                        if let GossipNetworkMessage::Protocol(protocol_message) = message {
+                            self0.send_signed_dkg_message(recipient, protocol_message)
+                        } else {
+                            self0.logger.error(
+                                "Received a non-protocol message from the controller".to_string(),
+                            );
+                        }
+                    }
+                    ToHandler::Gossip(v) => {
+                        if let GossipNetworkMessage::Protocol(protocol_message) = v {
+                            self0.gossip_dkg_signed_message(protocol_message);
+                        } else {
+                            self0.logger.error(
+                                "Received a non-protocol message from the controller".to_string(),
+                            );
+                        }
+                    }
+                }
+            }
+        });
+
+        let self2 = self.clone();
+        // second task, handles the incoming messages/events from the network stream.
+        let network_events_task = tokio::spawn(async move {
+            while let Some(event) = event_stream.next().await {
+                self2.handle_network_event(event).await;
+            }
+        });
+        // third task, handles the incoming events from the sync stream.
+        let mut self3 = self.clone();
+        let sync_events_task = tokio::spawn(async move {
+            while let Some(event) = sync_event_stream.next().await {
+                self3.handle_sync_event(event).await;
+            }
+        });
+        // wait for the first task to finish or error out.
+        //
+        // The reason why we wait for the first one to finish, is that it is a critical error
+        // if any of them finished before the others, the must either run all together or none.
+        //
+        // Here is more information about the reason why we do this:
+        // 1. if the network events task finished, it means that the network service has been
+        //   dropped, which means that the node is shutting down, or a network issue has occurred,
+        //   which is in this point a critical error.
+        // 2. if the incoming messages task finished, it means that the controller has been dropped,
+        //   which also could mean that there is now no communication between the controller and the
+        //   handler task, which is a critical error.
+        //   This could also mean that the node is shutting down, but in this case the network
+        // events   task should have finished as well.
+        // 3. The timer task, however, will never finish, unless the node is shutting down, in which
+        //  case the network events task should have finished as well.
+        let _result = futures::future::select_all(vec![
+            network_events_task,
+            incoming_messages_task,
+            sync_events_task,
+        ])
+        .await;
+        self.logger
+            .error("The DKG Gossip Handler has finished!!".to_string());
+    }
+
+    async fn handle_sync_event(&mut self, event: SyncEvent) {
+        match event {
+            SyncEvent::PeerConnected(remote) => {
+                let addr = iter::once(multiaddr::Protocol::P2p(remote.into()))
+                    .collect::<multiaddr::Multiaddr>();
+                let result = self
+                    .service
+                    .add_peers_to_reserved_set(self.protocol_name.clone(), HashSet::from([addr]));
+                if let Err(err) = result {
+                    self.logger
+                        .error(format!("Add reserved peer failed: {err}"));
+                }
+            }
+            SyncEvent::PeerDisconnected(remote) => {
+                if let Err(err) = self.service.remove_peers_from_reserved_set(
+                    self.protocol_name.clone(),
+                    iter::once(remote).collect(),
+                ) {
+                    self.logger
+                        .error(format!("Remove reserved peer failed: {err}"));
+                }
+            }
+        }
+    }
+
+    async fn handle_network_event(&self, event: Event) {
+        match event {
+            Event::Dht(_) => {}
+            Event::NotificationStreamOpened {
+                remote, protocol, ..
+            } if protocol == self.protocol_name => {
+                self.logger
+                    .debug(format!("Peer {remote} connected to gossip protocol"));
+                // Send our Handshake message to that peer.
+                self.send_handshake_message(remote).await;
+                let mut lock = self.peers.write();
+                let _was_in = lock.insert(
+                    remote,
+                    Peer {
+                        known_messages: LruHashSet::new(
+                            NonZeroUsize::new(MAX_KNOWN_MESSAGES).expect("Constant is nonzero"),
+                        ),
+                        message_counter: LruHashMap::new(
+                            NonZeroUsize::new(MAX_KNOWN_MESSAGES).expect("Constant is nonzero"),
+                        ),
+                        // At this point we don't know the authority id of the peer, so we set it to
+                        // None. We will update it once we receive the handshake message from that
+                        // peer.
+                        authority_id: None,
+                    },
+                );
+                debug_assert!(_was_in.is_none());
+            }
+            Event::NotificationStreamClosed { remote, protocol }
+                if protocol == self.protocol_name =>
+            {
+                let mut lock = self.peers.write();
+                let peer = lock.remove(&remote);
+                debug_assert!(peer.is_some());
+                // remove that peer's authority id from the authority id to peer id map (if any).
+                match peer {
+                    Some(peer) => {
+                        if let Some(authority_id) = peer.authority_id {
+                            let expected_remote =
+                                self.authority_id_to_peer_id.write().remove(&authority_id);
+                            // This should always be valid, if it isn't, it means that we have a bug
+                            // and somehow we have two peers with the same authority id.
+                            debug_assert_eq!(expected_remote, Some(remote));
+                        }
+                    }
+                    None => {
+                        self.logger.debug(format!(
+                            "Peer {remote} disconnected without sending handshake message"
+                        ));
+                    }
+                }
+                self.logger
+                    .debug(format!("Peer {remote} disconnected from gossip protocol"));
+            }
+
+            Event::NotificationsReceived { remote, messages } => {
+                for (protocol, message) in messages {
+                    if protocol != self.protocol_name {
+                        continue;
+                    }
+                    self.logger
+                        .debug(format!("Received message from {remote} from gossiping"));
+                    let maybe_dkg_network_message =
+                        bincode2::deserialize::<GossipNetworkMessage>(&message);
+                    let m = match maybe_dkg_network_message {
+                        Ok(m) => m,
+                        Err(e) => {
+                            self.logger.warn(format!("Failed to decode DKG Network message from peer {remote} with error: {e:?}"));
+                            self.service.report_peer(remote, rep::UNEXPECTED_MESSAGE);
+                            return;
+                        }
+                    };
+                    match m {
+                        GossipNetworkMessage::Handshake(h) => {
+                            self.on_handshake_message(remote, h).await
+                        }
+                        GossipNetworkMessage::Protocol(s) => {
+                            self.on_signed_dkg_message(remote, s).await
+                        }
+                    };
+                }
+            }
+            Event::NotificationStreamOpened { .. } => {}
+            Event::NotificationStreamClosed { .. } => {}
+        }
+    }
+
+    /// Creates and sends handshake message to the peer.
+    async fn send_handshake_message(&self, to_who: PeerId) {
+        let handshake_message = GossipNetworkMessage::Handshake(HandshakeMessage {
+            account_id: self.account_id,
+        });
+        let msg = bincode2::serialize(&handshake_message).expect("Failed to serialize message");
+        self.service
+            .write_notification(to_who, self.protocol_name.clone(), msg);
+    }
+
+    async fn on_handshake_message(&self, who: PeerId, message: HandshakeMessage) {
+        self.logger.debug(format!(
+            "Peer {who} is now connected as {}",
+            message.account_id
+        ));
+        let mut lock = self.peers.write();
+        if let Some(peer) = lock.get_mut(&who) {
+            peer.authority_id = Some(message.account_id);
+        } else {
+            self.logger.warn(format!(
+                "Peer {who} is not connected, but sent us a handshake message!!"
+            ));
+        }
+        self.authority_id_to_peer_id
+            .write()
+            .insert(message.account_id, who);
+    }
+
+    /// Called when peer sends us new signed DKG message.
+    async fn on_signed_dkg_message(&self, who: PeerId, message: GadgetProtocolMessage) {
+        // Check behavior of the peer.
+        self.logger.debug(format!(
+            "session {:?} | Received a signed DKG messages from {}",
+            message.associated_session_id, who,
+        ));
+
+        if let Some(_metrics) = self.metrics.as_ref() {
+            //metrics.dkg_signed_messages.inc();
+        }
+
+        if let Some(ref mut peer) = self.peers.write().get_mut(&who) {
+            let message_hash = message.hash();
+            peer.known_messages.insert(message_hash.clone());
+            let mut pending_messages_peers = self.pending_messages_peers.write();
+            let send_the_message = |message: GadgetProtocolMessage| {
+                if let Err(e) = self
+                    .to_receiver
+                    .send(GossipNetworkMessage::Protocol(message))
+                {
+                    self.logger.error(format!(
+                        "Failed to send message notification to DKG controller: {e:?}"
+                    ));
+                } else {
+                    self.logger.debug(
+                        "Message Notification sent to {recv_count} DKG controller listeners",
+                    );
+                }
+            };
+            match pending_messages_peers.inner.entry(message_hash.clone()) {
+                linked_hash_map::Entry::Vacant(entry) => {
+                    self.logger.debug(format!("NEW DKG MESSAGE FROM {who}"));
+                    if let Some(_metrics) = self.metrics.as_ref() {
+                        //metrics.dkg_new_signed_messages.inc();
+                    }
+
+                    send_the_message(message.clone());
+                    entry.insert(HashSet::from([who]));
+                    // This good, this peer is good, they sent us a message we didn't know about.
+                    // we should add some good reputation to them.
+                    self.service.report_peer(who, rep::GOOD_MESSAGE);
+                }
+                linked_hash_map::Entry::Occupied(mut entry) => {
+                    self.logger.debug(format!("OLD DKG MESSAGE FROM {who}"));
+                    if let Some(_metrics) = self.metrics.as_ref() {
+                        //metrics.dkg_old_signed_messages.inc();
+                    }
+                    // if we are here, that means this peer sent us a message we already know.
+                    let inserted = entry.get_mut().insert(who);
+                    // and if inserted is `false` that means this peer was already in the set
+                    // hence this not the first time we received this message from the exact same
+                    // peer.
+                    if !inserted {
+                        // we will increment the counter for this message.
+                        let old = peer
+                            .message_counter
+                            .get(&message_hash)
+                            .cloned()
+                            .unwrap_or(0);
+                        peer.message_counter.insert(message_hash, old + 1);
+                        // and if we have received this message from the same peer more than
+                        // `MAX_DUPLICATED_MESSAGES_PER_PEER` times, we should report this peer
+                        // as malicious.
+                        if old >= MAX_DUPLICATED_MESSAGES_PER_PEER {
+                            self.logger.warn(format!(
+                                "Reporting peer {who} as they are sending us the same message over and over again"
+                            ));
+                            self.service.report_peer(who, rep::DUPLICATE_MESSAGE);
+                        }
+                    }
+
+                    // send the old message anyways
+                    send_the_message(message.clone());
+                }
+            }
+        }
+
+        // if the gossip is enabled, we send the message to the gossiping peers
+        if self.gossip_enabled.load(Ordering::Relaxed) {
+            self.gossip_dkg_signed_message(message);
+        }
+    }
+
+    pub fn send_signed_dkg_message(&self, to_who: PeerId, message: GadgetProtocolMessage) {
+        let message_hash = message.hash();
+        if let Some(ref mut peer) = self.peers.write().get_mut(&to_who) {
+            let new_to_them = peer.known_messages.insert(message_hash);
+            if !new_to_them {
+                return;
+            }
+            let message = GossipNetworkMessage::Protocol(message);
+            let msg = bincode2::serialize(&message).expect("Failed to serialize message");
+            self.service
+                .write_notification(to_who, self.protocol_name.clone(), msg);
+
+            if let Some(_metrics) = self.metrics.as_ref() {
+                //metrics.dkg_propagated_messages.inc();
+            }
+        } else {
+            self.logger
+                .debug(format!("Peer {to_who} does not exist in known peers"));
+        }
+    }
+
+    fn gossip_dkg_signed_message(&self, message: GadgetProtocolMessage) {
+        // Check if the message has a recipient
+        let maybe_peer_id = match &message.to_account_id {
+            Some(recipient_id) => self
+                .authority_id_to_peer_id
+                .read()
+                .get(recipient_id)
+                .cloned(),
+            None => None,
+        };
+        // If we have a peer id, we send the message to that peer directly.
+        if let Some(peer_id) = maybe_peer_id {
+            self.logger
+                .debug(format!("Sending message to recipient {peer_id} using p2p"));
+            self.send_signed_dkg_message(peer_id, message.clone());
+            // potential bug "fix"
+            //return
+        } else if let Some(recipient_id) = &message.to {
+            self.logger.debug(format!(
+                "No direct connection to {recipient_id}, falling back to gossiping"
+            ));
+        } else {
+            self.logger
+                .debug("No specific recipient, broadcasting message to all peers".to_string());
+        }
+        // Otherwise, we fallback to sending the message to all peers.
+        let peer_ids = {
+            let peers_map = self.peers.read();
+            peers_map.keys().cloned().collect::<Vec<_>>()
+        };
+        if peer_ids.is_empty() {
+            let message_hash = message.hash();
+            self.logger
+                .warn(format!("No peers to gossip message {message_hash:?}"));
+            return;
+        }
+        for peer in peer_ids {
+            self.send_signed_dkg_message(peer, message.clone());
+        }
+    }
+}
+
+/// Wrapper around `LinkedHashMap` with bounded growth.
+///
+/// In the limit, for each element inserted the oldest existing element will be removed.
+#[derive(Debug, Clone)]
+pub struct LruHashMap<K: Hash + Eq, V> {
+    inner: LinkedHashMap<K, V>,
+    limit: NonZeroUsize,
+}
+
+impl<K: Hash + Eq, V> LruHashMap<K, V> {
+    /// Create a new `LruHashMap` with the given (exclusive) limit.
+    pub fn new(limit: NonZeroUsize) -> Self {
+        Self {
+            inner: LinkedHashMap::new(),
+            limit,
+        }
+    }
+
+    /// Insert element into the map.
+    ///
+    /// Returns `true` if this is a new element to the map, `false` otherwise.
+    /// Maintains the limit of the map by removing the oldest entry if necessary.
+    /// Inserting the same element will update its LRU position.
+    pub fn insert(&mut self, k: K, v: V) -> bool {
+        let is_new = self.inner.insert(k, v).is_none();
+        if self.inner.len() >= usize::from(self.limit) {
+            // Remove oldest entry
+            self.inner.pop_front();
+        }
+
+        is_new
+    }
+
+    /// Get an element from the map.
+    /// Returns `None` if the element is not in the map.
+    pub fn get(&self, k: &K) -> Option<&V> {
+        self.inner.get(k)
+    }
+}
+
+/// Wrapper around `LruHashMap` with bounded growth.
+///
+/// In the limit, for each element inserted the oldest existing element will be removed.
+#[derive(Debug, Clone)]
+pub struct LruHashSet<T: Hash + Eq> {
+    set: LruHashMap<T, ()>,
+}
+
+impl<T: Hash + Eq> LruHashSet<T> {
+    /// Create a new `LruHashSet` with the given (exclusive) limit.
+    pub fn new(limit: NonZeroUsize) -> Self {
+        Self {
+            set: LruHashMap::new(limit),
+        }
+    }
+
+    /// Insert element into the set.
+    ///
+    /// Returns `true` if this is a new element to the set, `false` otherwise.
+    /// Maintains the limit of the set by removing the oldest entry if necessary.
+    /// Inserting the same element will update its LRU position.
+    pub fn insert(&mut self, e: T) -> bool {
+        self.set.insert(e, ())
+    }
+}
diff --git a/protocols/mp-ecdsa/src/protocols/keygen.rs b/protocols/mp-ecdsa/src/protocols/keygen.rs
new file mode 100644
index 000000000..f32cd76e3
--- /dev/null
+++ b/protocols/mp-ecdsa/src/protocols/keygen.rs
@@ -0,0 +1,382 @@
+use crate::protocols::state_machine::{CurrentRoundBlame, StateMachineWrapper};
+use crate::protocols::util;
+use crate::protocols::util::PublicKeyGossipMessage;
+use crate::MpEcdsaProtocolConfig;
+use async_trait::async_trait;
+use curv::elliptic::curves::Secp256k1;
+use gadget_common::client::{AccountId, ClientWithApi, MpEcdsaClient};
+use gadget_common::debug_logger::DebugLogger;
+use gadget_common::gadget::message::GadgetProtocolMessage;
+use gadget_common::gadget::network::Network;
+use gadget_common::gadget::work_manager::WebbWorkManager;
+use gadget_common::gadget::{Job, WebbGadgetProtocol, WorkManagerConfig};
+use gadget_common::keystore::KeystoreBackend;
+use gadget_common::protocol::AsyncProtocol;
+use gadget_common::{Block, BlockImportNotification, FinalityNotification};
+use gadget_core::job::{BuiltExecutableJobWrapper, JobBuilder, JobError};
+use gadget_core::job_manager::{ProtocolWorkManager, WorkManagerInterface};
+use itertools::Itertools;
+use multi_party_ecdsa::gg_2020::state_machine::keygen::{Keygen, LocalKey};
+use pallet_jobs_rpc_runtime_api::JobsApi;
+use round_based::async_runtime::watcher::StderrWatcher;
+use round_based::{Msg, StateMachine};
+use sc_client_api::Backend;
+use sp_api::ProvideRuntimeApi;
+use sp_application_crypto::sp_core::keccak_256;
+use sp_application_crypto::RuntimePublic;
+use sp_core::crypto::KeyTypeId;
+use std::collections::{BTreeMap, HashMap};
+use std::error::Error;
+use std::sync::Arc;
+use tangle_primitives::jobs::{DKGResult, DkgKeyType, JobId, JobKey, JobResult, JobType};
+use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
+use tokio::sync::RwLock;
+
+pub struct MpEcdsaKeygenProtocol<B: Block, BE, KBE: KeystoreBackend, C, N> {
+    client: MpEcdsaClient<B, BE, KBE, C>,
+    network: N,
+    round_blames: Arc<
+        RwLock<
+            HashMap<
+                <WebbWorkManager as WorkManagerInterface>::TaskID,
+                tokio::sync::watch::Receiver<CurrentRoundBlame>,
+            >,
+        >,
+    >,
+    logger: DebugLogger,
+    id: sp_core::ecdsa::Public,
+    account_id: AccountId,
+}
+
+pub async fn create_protocol<B, BE, KBE, C, N>(
+    config: &MpEcdsaProtocolConfig,
+    client: MpEcdsaClient<B, BE, KBE, C>,
+    network: N,
+    logger: DebugLogger,
+) -> Result<MpEcdsaKeygenProtocol<B, BE, KBE, C, N>, Box<dyn Error>>
+where
+    B: Block,
+    BE: Backend<B>,
+    C: ClientWithApi<B, BE>,
+    KBE: KeystoreBackend,
+    N: Network,
+    <C as ProvideRuntimeApi<B>>::Api: JobsApi<B, AccountId>,
+{
+    Ok(MpEcdsaKeygenProtocol {
+        client,
+        network,
+        round_blames: Arc::new(RwLock::new(HashMap::new())),
+        logger,
+        id: config.id,
+        account_id: config.account_id,
+    })
+}
+
+#[async_trait]
+impl<
+        B: Block,
+        BE: Backend<B> + 'static,
+        C: ClientWithApi<B, BE>,
+        KBE: KeystoreBackend,
+        N: Network,
+    > WebbGadgetProtocol<B> for MpEcdsaKeygenProtocol<B, BE, KBE, C, N>
+where
+    <C as ProvideRuntimeApi<B>>::Api: JobsApi<B, AccountId>,
+{
+    async fn get_next_jobs(
+        &self,
+        notification: &FinalityNotification<B>,
+        now: u64,
+        job_manager: &ProtocolWorkManager<WebbWorkManager>,
+    ) -> Result<Option<Vec<Job>>, gadget_common::Error> {
+        let jobs = self
+            .client
+            .query_jobs_by_validator(notification.hash, self.account_id)
+            .await?
+            .into_iter()
+            .filter(|r| matches!(r.job_type, JobType::DKGTSSPhaseOne(..)));
+
+        let mut ret = vec![];
+
+        for job in jobs {
+            let participants = job
+                .job_type
+                .clone()
+                .get_participants()
+                .expect("Should exist for DKG");
+            if participants.contains(&self.account_id) {
+                let task_id = job.job_id.to_be_bytes();
+                let task_id = keccak_256(&task_id);
+                let session_id = 0; // We are not interested in sessions for the ECDSA protocol
+                let retry_id = job_manager
+                    .latest_retry_id(&task_id)
+                    .map(|r| r + 1)
+                    .unwrap_or(0);
+
+                let additional_params = MpEcdsaKeygenExtraParams {
+                    i: participants
+                        .iter()
+                        .position(|p| p == &self.account_id)
+                        .expect("Should exist") as u16,
+                    t: job.threshold.expect("T should exist for DKG") as u16,
+                    n: participants.len() as u16,
+                    job_id: job.job_id,
+                    job_key: job.job_type.get_job_key(),
+                };
+
+                let job = self
+                    .create(session_id, now, retry_id, task_id, additional_params)
+                    .await?;
+
+                ret.push(job);
+            }
+        }
+
+        Ok(Some(ret))
+    }
+
+    async fn process_block_import_notification(
+        &self,
+        _notification: BlockImportNotification<B>,
+        _job_manager: &ProtocolWorkManager<WebbWorkManager>,
+    ) -> Result<(), gadget_common::Error> {
+        Ok(())
+    }
+
+    async fn process_error(
+        &self,
+        error: gadget_common::Error,
+        _job_manager: &ProtocolWorkManager<WebbWorkManager>,
+    ) {
+        log::error!(target: "mp-ecdsa", "Error: {error:?}");
+    }
+
+    fn get_work_manager_config(&self) -> WorkManagerConfig {
+        WorkManagerConfig {
+            interval: None, // Manual polling
+            max_active_tasks: crate::constants::keygen_worker::MAX_RUNNING_TASKS,
+            max_pending_tasks: crate::constants::keygen_worker::MAX_ENQUEUED_TASKS,
+        }
+    }
+}
+
+pub struct MpEcdsaKeygenExtraParams {
+    i: u16,
+    t: u16,
+    n: u16,
+    job_id: JobId,
+    job_key: JobKey,
+}
+
+#[async_trait]
+impl<
+        B: Block,
+        BE: Backend<B> + 'static,
+        KBE: KeystoreBackend,
+        C: ClientWithApi<B, BE>,
+        N: Network,
+    > AsyncProtocol for MpEcdsaKeygenProtocol<B, BE, KBE, C, N>
+where
+    <C as ProvideRuntimeApi<B>>::Api: JobsApi<B, AccountId>,
+{
+    type AdditionalParams = MpEcdsaKeygenExtraParams;
+    async fn generate_protocol_from(
+        &self,
+        associated_block_id: <WebbWorkManager as WorkManagerInterface>::Clock,
+        associated_retry_id: <WebbWorkManager as WorkManagerInterface>::RetryID,
+        associated_session_id: <WebbWorkManager as WorkManagerInterface>::SessionID,
+        associated_task_id: <WebbWorkManager as WorkManagerInterface>::TaskID,
+        protocol_message_rx: UnboundedReceiver<GadgetProtocolMessage>,
+        additional_params: Self::AdditionalParams,
+    ) -> Result<BuiltExecutableJobWrapper, JobError> {
+        let key_store = self.client.key_store.clone();
+        let blame = self.round_blames.clone();
+        let protocol_output = Arc::new(tokio::sync::Mutex::new(None));
+        let protocol_output_clone = protocol_output.clone();
+        let client = self.client.clone();
+        let id = self.id;
+        let logger = self.logger.clone();
+        let logger_clone = logger.clone();
+        let round_blames = self.round_blames.clone();
+        let network = self.network.clone();
+
+        let (i, t, n) = (
+            additional_params.i,
+            additional_params.t,
+            additional_params.n,
+        );
+
+        Ok(JobBuilder::new()
+            .protocol(async move {
+                let keygen = Keygen::new(i, t, n).map_err(|err| JobError {
+                    reason: format!("Keygen setup error: {err:?}"),
+                })?;
+                let (current_round_blame_tx, current_round_blame_rx) =
+                    tokio::sync::watch::channel(CurrentRoundBlame::default());
+
+                round_blames
+                    .write()
+                    .await
+                    .insert(associated_task_id, current_round_blame_rx);
+
+                let (
+                    keygen_tx_to_outbound,
+                    keygen_rx_async_proto,
+                    broadcast_tx_to_outbound,
+                    broadcast_rx_from_gadget,
+                ) = util::create_job_manager_to_async_protocol_channel_split::<
+                    _,
+                    Msg<<Keygen as StateMachine>::MessageBody>,
+                    PublicKeyGossipMessage,
+                >(
+                    protocol_message_rx,
+                    associated_block_id,
+                    associated_retry_id,
+                    associated_session_id,
+                    associated_task_id,
+                    network,
+                );
+
+                let state_machine_wrapper =
+                    StateMachineWrapper::new(keygen, current_round_blame_tx, logger.clone());
+                let local_key = round_based::AsyncProtocol::new(
+                    state_machine_wrapper,
+                    keygen_rx_async_proto,
+                    keygen_tx_to_outbound,
+                )
+                .set_watcher(StderrWatcher)
+                .run()
+                .await
+                .map_err(|err| JobError {
+                    reason: format!("Keygen protocol error: {err:?}"),
+                })?;
+
+                let job_result = handle_public_key_gossip(
+                    &logger,
+                    &local_key,
+                    t,
+                    i,
+                    id,
+                    broadcast_tx_to_outbound,
+                    broadcast_rx_from_gadget,
+                )
+                .await?;
+
+                *protocol_output.lock().await = Some((local_key, job_result));
+                Ok(())
+            })
+            .post(async move {
+                // Check to see if there is any blame at the end of the protocol
+                if let Some(blame) = blame.write().await.remove(&associated_task_id) {
+                    let blame = blame.borrow();
+                    if !blame.blamed_parties.is_empty() {
+                        logger_clone.error(format!("Blame: {blame:?}"));
+                        return Err(JobError {
+                            reason: format!("Keygen blame: {blame:?}"),
+                        });
+                    }
+                }
+
+                // Store the keys locally, as well as submitting them to the blockchain
+                if let Some((local_key, job_result)) = protocol_output_clone.lock().await.take() {
+                    key_store
+                        .set(additional_params.job_id, local_key)
+                        .await
+                        .map_err(|err| JobError {
+                            reason: format!("Failed to store key: {err:?}"),
+                        })?;
+
+                    client
+                        .submit_job_result(
+                            additional_params.job_key,
+                            additional_params.job_id,
+                            job_result,
+                        )
+                        .await
+                        .map_err(|err| JobError {
+                            reason: format!("Failed to submit job result: {err:?}"),
+                        })?;
+                }
+
+                Ok(())
+            })
+            .build())
+    }
+}
+
+async fn handle_public_key_gossip(
+    logger: &DebugLogger,
+    local_key: &LocalKey<Secp256k1>,
+    threshold: u16,
+    i: u16,
+    my_id: sp_core::ecdsa::Public,
+    broadcast_tx_to_outbound: UnboundedSender<PublicKeyGossipMessage>,
+    mut broadcast_rx_from_gadget: UnboundedReceiver<PublicKeyGossipMessage>,
+) -> Result<JobResult, JobError> {
+    let serialized_public_key = local_key.public_key().to_bytes(true).to_vec();
+    // Sign the public key with our public key
+    let signature = my_id
+        .sign(KeyTypeId::default(), &serialized_public_key)
+        .ok_or_else(|| JobError {
+            reason: "Failed to sign public key".to_string(),
+        })?
+        .0
+        .to_vec();
+
+    let mut received_keys = BTreeMap::new();
+    received_keys.insert(i, signature.clone());
+    let mut received_participants = BTreeMap::new();
+    received_participants.insert(i, my_id);
+
+    broadcast_tx_to_outbound
+        .send(PublicKeyGossipMessage {
+            from: i as _,
+            to: None,
+            signature,
+            id: my_id,
+        })
+        .map_err(|err| JobError {
+            reason: format!("Failed to send public key: {err:?}"),
+        })?;
+
+    for _ in 0..threshold {
+        let message: PublicKeyGossipMessage =
+            broadcast_rx_from_gadget
+                .recv()
+                .await
+                .ok_or_else(|| JobError {
+                    reason: "Failed to receive public key".to_string(),
+                })?;
+
+        let from = message.from;
+        if received_keys.contains_key(&(from as u16)) {
+            logger.warn("Received duplicate key");
+            continue;
+        }
+
+        received_keys.insert(from as u16, message.signature);
+
+        received_participants.insert(from as u16, message.id);
+    }
+
+    // Order and collect the map to ensure symmetric submission to blockchain
+    let signatures = received_keys
+        .into_iter()
+        .sorted_by_key(|x| x.0)
+        .map(|r| r.1)
+        .collect();
+
+    let participants = received_participants
+        .into_iter()
+        .sorted_by_key(|x| x.0)
+        .map(|r| r.1 .0.to_vec())
+        .collect();
+
+    Ok(JobResult::DKGPhaseOne(DKGResult {
+        key_type: DkgKeyType::Ecdsa,
+        key: serialized_public_key,
+        participants,
+        threshold: threshold as _,
+        signatures,
+    }))
+}
diff --git a/protocols/mp-ecdsa/src/protocols/mod.rs b/protocols/mp-ecdsa/src/protocols/mod.rs
new file mode 100644
index 000000000..4fe2dbfb7
--- /dev/null
+++ b/protocols/mp-ecdsa/src/protocols/mod.rs
@@ -0,0 +1,4 @@
+pub mod keygen;
+pub mod sign;
+pub mod state_machine;
+pub mod util;
diff --git a/protocols/mp-ecdsa/src/protocols/sign.rs b/protocols/mp-ecdsa/src/protocols/sign.rs
new file mode 100644
index 000000000..102203fe8
--- /dev/null
+++ b/protocols/mp-ecdsa/src/protocols/sign.rs
@@ -0,0 +1,436 @@
+use crate::protocols::state_machine::{CurrentRoundBlame, StateMachineWrapper};
+use crate::protocols::util;
+use crate::protocols::util::VotingMessage;
+use crate::MpEcdsaProtocolConfig;
+use async_trait::async_trait;
+use curv::arithmetic::Converter;
+use curv::elliptic::curves::Secp256k1;
+use curv::BigInt;
+use gadget_common::client::{AccountId, ClientWithApi, MpEcdsaClient};
+use gadget_common::debug_logger::DebugLogger;
+use gadget_common::gadget::message::{GadgetProtocolMessage, UserID};
+use gadget_common::gadget::network::Network;
+use gadget_common::gadget::work_manager::WebbWorkManager;
+use gadget_common::gadget::{Job, WebbGadgetProtocol, WorkManagerConfig};
+use gadget_common::keystore::KeystoreBackend;
+use gadget_common::protocol::AsyncProtocol;
+use gadget_common::{Block, BlockImportNotification, FinalityNotification};
+use gadget_core::job::{BuiltExecutableJobWrapper, JobBuilder, JobError};
+use gadget_core::job_manager::{ProtocolWorkManager, WorkManagerInterface};
+use multi_party_ecdsa::gg_2020::party_i::verify;
+use multi_party_ecdsa::gg_2020::state_machine::keygen::LocalKey;
+use multi_party_ecdsa::gg_2020::state_machine::sign::{
+    CompletedOfflineStage, OfflineStage, PartialSignature, SignManual,
+};
+use pallet_jobs_rpc_runtime_api::JobsApi;
+use parity_scale_codec::Encode;
+use round_based::async_runtime::watcher::StderrWatcher;
+use round_based::{Msg, StateMachine};
+use sc_client_api::Backend;
+use sp_api::ProvideRuntimeApi;
+use sp_core::ecdsa::Signature;
+use sp_core::keccak_256;
+use std::collections::HashMap;
+use std::error::Error;
+use std::sync::Arc;
+use tangle_primitives::jobs::{DKGSignatureResult, DkgKeyType, JobId, JobKey, JobResult, JobType};
+use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
+use tokio::sync::RwLock;
+
+pub struct MpEcdsaSigningProtocol<B: Block, BE, KBE: KeystoreBackend, C, N> {
+    client: MpEcdsaClient<B, BE, KBE, C>,
+    network: N,
+    round_blames: Arc<
+        RwLock<
+            HashMap<
+                <WebbWorkManager as WorkManagerInterface>::TaskID,
+                tokio::sync::watch::Receiver<CurrentRoundBlame>,
+            >,
+        >,
+    >,
+    logger: DebugLogger,
+    account_id: AccountId,
+}
+
+pub async fn create_protocol<B, BE, KBE, C, N>(
+    config: &MpEcdsaProtocolConfig,
+    logger: DebugLogger,
+    client: MpEcdsaClient<B, BE, KBE, C>,
+    network: N,
+) -> Result<MpEcdsaSigningProtocol<B, BE, KBE, C, N>, Box<dyn Error>>
+where
+    B: Block,
+    BE: Backend<B>,
+    C: ClientWithApi<B, BE>,
+    KBE: KeystoreBackend,
+    N: Network,
+    <C as ProvideRuntimeApi<B>>::Api: JobsApi<B, AccountId>,
+{
+    Ok(MpEcdsaSigningProtocol {
+        client,
+        network,
+        round_blames: Arc::new(Default::default()),
+        logger,
+        account_id: config.account_id,
+    })
+}
+
+#[async_trait]
+impl<
+        B: Block,
+        BE: Backend<B> + 'static,
+        C: ClientWithApi<B, BE>,
+        KBE: KeystoreBackend,
+        N: Network,
+    > WebbGadgetProtocol<B> for MpEcdsaSigningProtocol<B, BE, KBE, C, N>
+where
+    <C as ProvideRuntimeApi<B>>::Api: JobsApi<B, AccountId>,
+{
+    async fn get_next_jobs(
+        &self,
+        notification: &FinalityNotification<B>,
+        now: u64,
+        job_manager: &ProtocolWorkManager<WebbWorkManager>,
+    ) -> Result<Option<Vec<Job>>, gadget_common::Error> {
+        let jobs = self
+            .client
+            .query_jobs_by_validator(notification.hash, self.account_id)
+            .await?
+            .into_iter()
+            .filter(|r| matches!(r.job_type, JobType::DKGTSSPhaseTwo(..)));
+
+        let mut ret = vec![];
+
+        for job in jobs {
+            let participants = job.participants.expect("Should exist for stage 2 signing");
+            if participants.contains(&self.account_id) {
+                let task_id = job.job_id.to_be_bytes();
+                let task_id = keccak_256(&task_id);
+                let session_id = 0; // We are not interested in sessions for the ECDSA protocol
+                                    // For the retry ID, get the latest retry and increment by 1 to set the next retry ID
+                                    // If no retry ID exists, it means the job is new, thus, set the retry_id to 0
+                let retry_id = job_manager
+                    .latest_retry_id(&task_id)
+                    .map(|r| r + 1)
+                    .unwrap_or(0);
+
+                if let Some(key) = self
+                    .client
+                    .key_store
+                    .get(&job.job_id)
+                    .await
+                    .map_err(|err| gadget_common::Error::ClientError {
+                        err: err.to_string(),
+                    })?
+                {
+                    let input_data_to_sign = if let JobType::DKGTSSPhaseTwo(sig) = &job.job_type {
+                        sig.submission.clone()
+                    } else {
+                        panic!("Should be DKGSignature");
+                    };
+
+                    let additional_params = MpEcdsaSigningExtraParams {
+                        i: participants
+                            .iter()
+                            .position(|p| p == &self.account_id)
+                            .expect("Should exist") as u16,
+                        t: job.threshold.expect("T should exist for stage 2 signing") as u16,
+                        signers: (0..participants.len()).map(|r| r as u16).collect(),
+                        job_id: job.job_id,
+                        job_key: job.job_type.get_job_key(),
+                        key,
+                        input_data_to_sign,
+                    };
+
+                    let job = self
+                        .create(session_id, now, retry_id, task_id, additional_params)
+                        .await?;
+
+                    ret.push(job);
+                } else {
+                    self.logger.warn(format!(
+                        "No key found for job ID: {job_id:?}",
+                        job_id = job.job_id
+                    ));
+                }
+            }
+        }
+
+        Ok(Some(ret))
+    }
+
+    async fn process_block_import_notification(
+        &self,
+        _notification: BlockImportNotification<B>,
+        _job_manager: &ProtocolWorkManager<WebbWorkManager>,
+    ) -> Result<(), gadget_common::Error> {
+        Ok(())
+    }
+
+    async fn process_error(
+        &self,
+        error: gadget_common::Error,
+        _job_manager: &ProtocolWorkManager<WebbWorkManager>,
+    ) {
+        log::error!(target: "mp-ecdsa", "Error: {error:?}");
+    }
+
+    fn get_work_manager_config(&self) -> WorkManagerConfig {
+        WorkManagerConfig {
+            interval: Some(crate::constants::signing_worker::JOB_POLL_INTERVAL),
+            max_active_tasks: crate::constants::signing_worker::MAX_RUNNING_TASKS,
+            max_pending_tasks: crate::constants::signing_worker::MAX_ENQUEUED_TASKS,
+        }
+    }
+}
+
+pub struct MpEcdsaSigningExtraParams {
+    i: u16,
+    t: u16,
+    signers: Vec<u16>,
+    job_id: JobId,
+    job_key: JobKey,
+    key: LocalKey<Secp256k1>,
+    input_data_to_sign: Vec<u8>,
+}
+
+#[async_trait]
+impl<
+        B: Block,
+        BE: Backend<B> + 'static,
+        KBE: KeystoreBackend,
+        C: ClientWithApi<B, BE>,
+        N: Network,
+    > AsyncProtocol for MpEcdsaSigningProtocol<B, BE, KBE, C, N>
+where
+    <C as ProvideRuntimeApi<B>>::Api: JobsApi<B, AccountId>,
+{
+    type AdditionalParams = MpEcdsaSigningExtraParams;
+    async fn generate_protocol_from(
+        &self,
+        associated_block_id: <WebbWorkManager as WorkManagerInterface>::Clock,
+        associated_retry_id: <WebbWorkManager as WorkManagerInterface>::RetryID,
+        associated_session_id: <WebbWorkManager as WorkManagerInterface>::SessionID,
+        associated_task_id: <WebbWorkManager as WorkManagerInterface>::TaskID,
+        protocol_message_rx: UnboundedReceiver<GadgetProtocolMessage>,
+        additional_params: Self::AdditionalParams,
+    ) -> Result<BuiltExecutableJobWrapper, JobError> {
+        let blame = self.round_blames.clone();
+        let debug_logger_post = self.logger.clone();
+        let debug_logger_proto = debug_logger_post.clone();
+        let protocol_output = Arc::new(tokio::sync::Mutex::new(None));
+        let protocol_output_clone = protocol_output.clone();
+        let client = self.client.clone();
+        let round_blames = self.round_blames.clone();
+        let network = self.network.clone();
+
+        let (i, signers, t, key, input_data_to_sign) = (
+            additional_params.i,
+            additional_params.signers,
+            additional_params.t,
+            additional_params.key,
+            additional_params.input_data_to_sign.clone(),
+        );
+
+        let key_clone = key.clone();
+
+        Ok(JobBuilder::new()
+            .protocol(async move {
+                let offline_i = get_offline_i(i, &signers);
+
+                let signing =
+                    OfflineStage::new(offline_i, signers, key).map_err(|err| JobError {
+                        reason: format!("Failed to create offline stage: {err:?}"),
+                    })?;
+                let (current_round_blame_tx, current_round_blame_rx) =
+                    tokio::sync::watch::channel(CurrentRoundBlame::default());
+
+                round_blames
+                    .write()
+                    .await
+                    .insert(associated_task_id, current_round_blame_rx);
+
+                let (
+                    tx_to_outbound_offline,
+                    rx_async_proto_offline,
+                    tx_to_outbound_voting,
+                    rx_async_proto_voting,
+                ) = util::create_job_manager_to_async_protocol_channel_split::<
+                    _,
+                    Msg<<OfflineStage as StateMachine>::MessageBody>,
+                    VotingMessage,
+                >(
+                    protocol_message_rx,
+                    associated_block_id,
+                    associated_retry_id,
+                    associated_session_id,
+                    associated_task_id,
+                    network,
+                );
+
+                let state_machine_wrapper = StateMachineWrapper::new(
+                    signing,
+                    current_round_blame_tx,
+                    debug_logger_proto.clone(),
+                );
+                let completed_offline_stage = round_based::AsyncProtocol::new(
+                    state_machine_wrapper,
+                    rx_async_proto_offline,
+                    tx_to_outbound_offline,
+                )
+                .set_watcher(StderrWatcher)
+                .run()
+                .await
+                .map_err(|err| JobError {
+                    reason: format!("Keygen protocol error: {err:?}"),
+                })?;
+
+                debug_logger_proto.info(format!(
+                    "*** Completed offline stage: {:?}",
+                    completed_offline_stage.public_key()
+                ));
+
+                // We will sign over the unique task ID
+                let message = BigInt::from_bytes(&input_data_to_sign);
+
+                // Conclude with the voting stage
+                let signature = voting_stage(
+                    offline_i,
+                    t,
+                    message,
+                    completed_offline_stage,
+                    rx_async_proto_voting,
+                    tx_to_outbound_voting,
+                    &debug_logger_proto,
+                )
+                .await?;
+                *protocol_output.lock().await = Some(signature);
+                Ok(())
+            })
+            .post(async move {
+                // Check to see if there is any blame at the end of the protocol
+                if let Some(blame) = blame.write().await.remove(&associated_task_id) {
+                    let blame = blame.borrow();
+                    if !blame.blamed_parties.is_empty() {
+                        debug_logger_post.error(format!("Blame: {blame:?}"));
+                        return Err(JobError {
+                            reason: format!("Signing blame: {blame:?}"),
+                        });
+                    }
+                }
+
+                // Submit the protocol output to the blockchain
+                if let Some(signature) = protocol_output_clone.lock().await.take() {
+                    let signature: Vec<u8> = signature.encode();
+                    let job_result = JobResult::DKGPhaseTwo(DKGSignatureResult {
+                        key_type: DkgKeyType::Ecdsa,
+                        data: additional_params.input_data_to_sign,
+                        signature,
+                        signing_key: key_clone.public_key().to_bytes(true).to_vec(),
+                    });
+
+                    client
+                        .submit_job_result(
+                            additional_params.job_key,
+                            additional_params.job_id,
+                            job_result,
+                        )
+                        .await
+                        .map_err(|err| JobError {
+                            reason: format!("Failed to submit job result: {err:?}"),
+                        })?;
+                }
+
+                Ok(())
+            })
+            .build())
+    }
+}
+
+/// Finds our index inside the list of signers, then adds 1 since the Offline state machine
+/// expects the index to start from 1.
+fn get_offline_i(i: u16, signers: &[u16]) -> u16 {
+    (signers.iter().position(|s| s == &i).expect("Should exist") + 1) as u16
+}
+
+async fn voting_stage(
+    offline_i: u16,
+    threshold: u16,
+    message: BigInt,
+    completed_offline_stage: CompletedOfflineStage,
+    mut msg_rx: UnboundedReceiver<VotingMessage>,
+    msg_tx: UnboundedSender<VotingMessage>,
+    debug_logger: &DebugLogger,
+) -> Result<Signature, JobError> {
+    let offline_stage_pub_key = completed_offline_stage.public_key().clone();
+    let (signing, partial_signature) = SignManual::new(message.clone(), completed_offline_stage)
+        .map_err(|err| JobError {
+            reason: format!("Failed to create voting stage: {err:?}"),
+        })?;
+
+    let partial_sig_bytes = bincode2::serialize(&partial_signature).map_err(|err| JobError {
+        reason: format!("Failed to serialize partial signature: {err:?}"),
+    })?;
+
+    let payload = VotingMessage {
+        from: offline_i as UserID,
+        to: None, // Broadcast to everyone
+        payload: partial_sig_bytes,
+    };
+
+    msg_tx.send(payload).map_err(|err| JobError {
+        reason: format!("Failed to send partial signature: {err:?}"),
+    })?;
+
+    let mut sigs = HashMap::with_capacity(threshold as _);
+
+    while let Some(vote_message) = msg_rx.recv().await {
+        let vote_message: VotingMessage = vote_message;
+        if sigs.contains_key(&vote_message.from) {
+            debug_logger.warn(format!(
+                "Received duplicate signature from {}",
+                vote_message.from
+            ));
+            continue;
+        }
+
+        if let Ok(p_sig) = bincode2::deserialize::<PartialSignature>(&vote_message.payload) {
+            sigs.insert(vote_message.from, p_sig);
+
+            if sigs.len() == threshold as usize {
+                break;
+            }
+        } else {
+            debug_logger.warn(format!(
+                "Received invalid signature bytes from {}",
+                vote_message.from
+            ));
+        }
+    }
+
+    if sigs.len() != threshold as usize {
+        return Err(JobError {
+            reason: format!(
+                "Failed to collect enough signatures: {}/{}",
+                sigs.len(),
+                threshold
+            ),
+        });
+    }
+
+    // Aggregate and complete the signature
+    let sigs: Vec<PartialSignature> = sigs.into_values().collect();
+    let signature = signing.complete(&sigs).map_err(|err| JobError {
+        reason: format!("Failed to complete signature: {err:?}"),
+    })?;
+
+    // Verify the signature
+    verify(&signature, &offline_stage_pub_key, &message).map_err(|err| JobError {
+        reason: format!("Failed to verify signature: {err:?}"),
+    })?;
+
+    // Convert the signature to a substrate-compatible format
+    crate::util::convert_signature(debug_logger, &signature).ok_or_else(|| JobError {
+        reason: "Failed to convert signature to Substrate-compatible format".to_string(),
+    })
+}
diff --git a/protocols/mp-ecdsa/src/protocols/state_machine.rs b/protocols/mp-ecdsa/src/protocols/state_machine.rs
new file mode 100644
index 000000000..4fe38666a
--- /dev/null
+++ b/protocols/mp-ecdsa/src/protocols/state_machine.rs
@@ -0,0 +1,160 @@
+use gadget_common::debug_logger::DebugLogger;
+use multi_party_ecdsa::gg_2020::state_machine::traits::RoundBlame;
+use multi_party_ecdsa::MessageRoundID;
+use round_based::{Msg, StateMachine};
+use sp_application_crypto::serde::Serialize;
+use std::collections::HashSet;
+use std::fmt::Debug;
+
+pub(crate) struct StateMachineWrapper<T> {
+    sm: T,
+    current_round_blame: tokio::sync::watch::Sender<CurrentRoundBlame>,
+    // stores a list of received messages
+    received_messages: HashSet<Vec<u8>>,
+    logger: DebugLogger,
+}
+
+impl<T: StateMachine + RoundBlame + Debug> StateMachineWrapper<T> {
+    pub fn new(
+        sm: T,
+        current_round_blame: tokio::sync::watch::Sender<CurrentRoundBlame>,
+        logger: DebugLogger,
+    ) -> Self {
+        Self {
+            sm,
+            current_round_blame,
+            logger,
+            received_messages: HashSet::new(),
+        }
+    }
+}
+
+impl<T> StateMachine for StateMachineWrapper<T>
+where
+    T: StateMachine + RoundBlame + Debug,
+    <T as StateMachine>::Err: Debug,
+    <T as StateMachine>::MessageBody: Serialize + MessageRoundID,
+{
+    type Err = T::Err;
+    type Output = T::Output;
+    type MessageBody = T::MessageBody;
+
+    fn handle_incoming(&mut self, msg: Msg<Self::MessageBody>) -> Result<(), Self::Err> {
+        let (round, sender, _receiver) = (msg.body.round_id(), msg.sender, msg.receiver);
+
+        self.logger.trace(format!(
+            "Handling incoming message round={}, sender={}",
+            round, sender
+        ));
+
+        if round < self.current_round() {
+            self.logger
+                .trace(format!("Message for round={round} is outdated, ignoring",));
+            return Ok(());
+        }
+
+        // Before passing to the state machine, make sure that we haven't already received the same
+        // message (this is needed as we use a gossiping protocol to send messages, and we don't
+        // want to process the same message twice)
+        let msg_serde = bincode2::serialize(&msg).expect("Failed to serialize message");
+        if !self.received_messages.insert(msg_serde) {
+            self.logger.trace(format!(
+                "Already received message for round={}, sender={}",
+                round, sender
+            ));
+            return Ok(());
+        }
+
+        let result = self.sm.handle_incoming(msg.clone());
+
+        if let Some(err) = result.as_ref().err() {
+            self.logger.error(format!("StateMachine error: {err:?}"));
+        }
+
+        // Get the round blame to update round blame
+        let round_blame = self.round_blame();
+
+        self.logger.trace(format!(
+            "SM After: {:?} || round_blame: {:?}",
+            &self.sm, round_blame
+        ));
+
+        result
+    }
+
+    fn message_queue(&mut self) -> &mut Vec<Msg<Self::MessageBody>> {
+        self.sm.message_queue()
+    }
+
+    fn wants_to_proceed(&self) -> bool {
+        self.sm.wants_to_proceed()
+    }
+
+    fn proceed(&mut self) -> Result<(), Self::Err> {
+        self.logger.trace(format!(
+            "Trying to proceed: current round ({:?}), waiting for msgs from parties: ({:?})",
+            self.current_round(),
+            self.round_blame(),
+        ));
+        let result = self.sm.proceed();
+
+        self.logger.trace(format!(
+            "Proceeded through SM. New current round ({:?}), waiting for msgs from parties: ({:?})",
+            self.current_round(),
+            self.round_blame(),
+        ));
+
+        result
+    }
+
+    fn round_timeout(&self) -> Option<std::time::Duration> {
+        self.sm.round_timeout()
+    }
+
+    fn round_timeout_reached(&mut self) -> Self::Err {
+        self.sm.round_timeout_reached()
+    }
+
+    fn is_finished(&self) -> bool {
+        self.sm.is_finished()
+    }
+
+    fn pick_output(&mut self) -> Option<Result<Self::Output, Self::Err>> {
+        self.sm.pick_output()
+    }
+
+    fn current_round(&self) -> u16 {
+        self.sm.current_round()
+    }
+
+    fn total_rounds(&self) -> Option<u16> {
+        self.sm.total_rounds()
+    }
+
+    fn party_ind(&self) -> u16 {
+        self.sm.party_ind()
+    }
+
+    fn parties(&self) -> u16 {
+        self.sm.parties()
+    }
+}
+
+impl<T: StateMachine + RoundBlame> RoundBlame for StateMachineWrapper<T> {
+    fn round_blame(&self) -> (u16, Vec<u16>) {
+        let (unreceived_messages, blamed_parties) = self.sm.round_blame();
+        self.logger
+            .debug(format!("Not received messages from : {blamed_parties:?}"));
+        let _ = self.current_round_blame.send(CurrentRoundBlame {
+            unreceived_messages,
+            blamed_parties: blamed_parties.clone(),
+        });
+        (unreceived_messages, blamed_parties)
+    }
+}
+
+#[derive(Default, Debug)]
+pub struct CurrentRoundBlame {
+    pub unreceived_messages: u16,
+    pub blamed_parties: Vec<u16>,
+}
diff --git a/protocols/mp-ecdsa/src/protocols/util.rs b/protocols/mp-ecdsa/src/protocols/util.rs
new file mode 100644
index 000000000..7f8599800
--- /dev/null
+++ b/protocols/mp-ecdsa/src/protocols/util.rs
@@ -0,0 +1,171 @@
+//! When delivering messages to an async protocol, we want o make sure we don't mix up voting and public key gossip messages
+//! Thus, this file contains a function that takes a channel from the gadget to the async protocol and splits it into two channels
+use futures::StreamExt;
+use gadget_common::gadget::message::{GadgetProtocolMessage, UserID};
+use gadget_common::gadget::network::Network;
+use gadget_common::gadget::work_manager::WebbWorkManager;
+use gadget_core::job_manager::WorkManagerInterface;
+use round_based::Msg;
+use serde::de::DeserializeOwned;
+use serde::{Deserialize, Serialize};
+use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
+
+#[derive(Serialize, Deserialize, Debug)]
+pub enum SplitChannelMessage<C1, C2> {
+    Channel1(C1),
+    Channel2(C2),
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct VotingMessage {
+    pub from: UserID,
+    pub to: Option<UserID>,
+    pub payload: Vec<u8>,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct PublicKeyGossipMessage {
+    pub from: UserID,
+    pub to: Option<UserID>,
+    pub signature: Vec<u8>,
+    pub id: sp_core::ecdsa::Public,
+}
+
+pub fn create_job_manager_to_async_protocol_channel_split<
+    N: Network + 'static,
+    C1: Serialize + DeserializeOwned + HasSenderAndReceiver + Send + 'static,
+    C2: Serialize + DeserializeOwned + HasSenderAndReceiver + Send + 'static,
+>(
+    mut rx_gadget: UnboundedReceiver<GadgetProtocolMessage>,
+    associated_block_id: <WebbWorkManager as WorkManagerInterface>::Clock,
+    associated_retry_id: <WebbWorkManager as WorkManagerInterface>::RetryID,
+    associated_session_id: <WebbWorkManager as WorkManagerInterface>::SessionID,
+    associated_task_id: <WebbWorkManager as WorkManagerInterface>::TaskID,
+    network: N,
+) -> (
+    futures::channel::mpsc::UnboundedSender<C1>,
+    futures::channel::mpsc::UnboundedReceiver<std::io::Result<C1>>,
+    UnboundedSender<C2>,
+    UnboundedReceiver<C2>,
+) {
+    let (tx_to_async_proto_1, rx_for_async_proto_1) = futures::channel::mpsc::unbounded();
+    let (tx_to_async_proto_2, rx_for_async_proto_2) = tokio::sync::mpsc::unbounded_channel();
+    // Take the messages from the gadget and send them to the async protocol
+    tokio::task::spawn(async move {
+        while let Some(msg) = rx_gadget.recv().await {
+            match bincode2::deserialize::<SplitChannelMessage<C1, C2>>(&msg.payload) {
+                Ok(msg) => match msg {
+                    SplitChannelMessage::Channel1(msg) => {
+                        if tx_to_async_proto_1.unbounded_send(Ok(msg)).is_err() {
+                            log::error!("Failed to send message to protocol");
+                        }
+                    }
+                    SplitChannelMessage::Channel2(msg) => {
+                        if tx_to_async_proto_2.send(msg).is_err() {
+                            log::error!("Failed to send message to protocol");
+                        }
+                    }
+                },
+                Err(err) => {
+                    log::error!("Failed to deserialize message: {err:?}");
+                }
+            }
+        }
+    });
+
+    let (tx_to_outbound_1, mut rx_to_outbound_1) = futures::channel::mpsc::unbounded::<C1>();
+    let (tx_to_outbound_2, mut rx_to_outbound_2) = tokio::sync::mpsc::unbounded_channel::<C2>();
+    let network_clone = network.clone();
+    // Take the messages the async protocol sends to the outbound channel and send them to the gadget
+    tokio::task::spawn(async move {
+        let offline_task = async move {
+            while let Some(msg) = rx_to_outbound_1.next().await {
+                let from = msg.sender();
+                let to = msg.receiver();
+                let msg = SplitChannelMessage::<C1, C2>::Channel1(msg);
+                let msg = GadgetProtocolMessage {
+                    associated_block_id,
+                    associated_session_id,
+                    associated_retry_id,
+                    task_hash: associated_task_id,
+                    from,
+                    to,
+                    payload: bincode2::serialize(&msg).expect("Failed to serialize message"),
+                    from_account_id: None, // TODO: Mapping of [task_hash] => UserID => AccountID for mapping userIDs to accountIDs
+                    to_account_id: None, // TODO: Mapping of [task_hash] => UserID => AccountID for mapping userIDs to accountIDs
+                };
+
+                if network.send_message(msg).await.is_err() {
+                    log::error!("Failed to send message to outbound");
+                }
+            }
+        };
+
+        let voting_task = async move {
+            while let Some(msg) = rx_to_outbound_2.recv().await {
+                let from = msg.sender();
+                let to = msg.receiver();
+                let msg = SplitChannelMessage::<C1, C2>::Channel2(msg);
+                let msg = GadgetProtocolMessage {
+                    associated_block_id,
+                    associated_session_id,
+                    associated_retry_id,
+                    task_hash: associated_task_id,
+                    from,
+                    to,
+                    payload: bincode2::serialize(&msg).expect("Failed to serialize message"),
+                    from_account_id: None, // TODO: Mapping of [task_hash] => UserID => AccountID for mapping userIDs to accountIDs
+                    to_account_id: None, // TODO: Mapping of [task_hash] => UserID => AccountID for mapping userIDs to accountIDs
+                };
+
+                if network_clone.send_message(msg).await.is_err() {
+                    log::error!("Failed to send message to outbound");
+                }
+            }
+        };
+
+        tokio::join!(offline_task, voting_task);
+    });
+
+    (
+        tx_to_outbound_1,
+        rx_for_async_proto_1,
+        tx_to_outbound_2,
+        rx_for_async_proto_2,
+    )
+}
+
+pub trait HasSenderAndReceiver {
+    fn sender(&self) -> UserID;
+    fn receiver(&self) -> Option<UserID>;
+}
+
+impl<T> HasSenderAndReceiver for Msg<T> {
+    fn sender(&self) -> UserID {
+        self.sender as UserID
+    }
+
+    fn receiver(&self) -> Option<UserID> {
+        self.receiver.map(|r| r as UserID)
+    }
+}
+
+impl HasSenderAndReceiver for VotingMessage {
+    fn sender(&self) -> UserID {
+        self.from
+    }
+
+    fn receiver(&self) -> Option<UserID> {
+        self.to
+    }
+}
+
+impl HasSenderAndReceiver for PublicKeyGossipMessage {
+    fn sender(&self) -> UserID {
+        self.from
+    }
+
+    fn receiver(&self) -> Option<UserID> {
+        self.to
+    }
+}
diff --git a/protocols/mp-ecdsa/src/util.rs b/protocols/mp-ecdsa/src/util.rs
new file mode 100644
index 000000000..ab8d32abb
--- /dev/null
+++ b/protocols/mp-ecdsa/src/util.rs
@@ -0,0 +1,40 @@
+use curv::arithmetic::Converter;
+use gadget_common::debug_logger::DebugLogger;
+use multi_party_ecdsa::gg_2020::party_i::SignatureRecid;
+use sp_core::ecdsa::Signature;
+
+pub fn convert_signature(
+    debug_logger: &DebugLogger,
+    sig_recid: &SignatureRecid,
+) -> Option<Signature> {
+    let r = sig_recid.r.to_bigint().to_bytes();
+    let s = sig_recid.s.to_bigint().to_bytes();
+    let v = sig_recid.recid + 27u8;
+
+    let mut sig_vec: Vec<u8> = Vec::new();
+
+    for _ in 0..(32 - r.len()) {
+        sig_vec.extend([0]);
+    }
+    sig_vec.extend_from_slice(&r);
+
+    for _ in 0..(32 - s.len()) {
+        sig_vec.extend([0]);
+    }
+    sig_vec.extend_from_slice(&s);
+
+    sig_vec.extend([v]);
+
+    if 65 != sig_vec.len() {
+        debug_logger.warn(format!(
+            "Invalid signature len: {}, expected 65",
+            sig_vec.len()
+        ));
+        return None;
+    }
+
+    let mut dkg_sig_arr: [u8; 65] = [0; 65];
+    dkg_sig_arr.copy_from_slice(&sig_vec[0..65]);
+
+    Some(Signature(dkg_sig_arr))
+}
diff --git a/rust-toolchain.toml b/rust-toolchain.toml
index 24f0d828a..516c9f1d5 100644
--- a/rust-toolchain.toml
+++ b/rust-toolchain.toml
@@ -1,4 +1,4 @@
 [toolchain]
-channel = "stable"
+channel = "nightly"
 components = ["rustfmt", "clippy", "rust-src"]
 targets = []
diff --git a/zk-gadget/Cargo.toml b/zk-gadget/Cargo.toml
index 629eb204c..d13d9d28f 100644
--- a/zk-gadget/Cargo.toml
+++ b/zk-gadget/Cargo.toml
@@ -8,7 +8,7 @@ edition = "2021"
 [dependencies]
 tokio-rustls = { workspace = true }
 mpc-net = { workspace = true }
-webb-gadget = { workspace = true }
+gadget-common = { workspace = true }
 gadget-core = { workspace = true }
 bincode2 = { workspace = true }
 tokio = { workspace = true }
@@ -20,6 +20,7 @@ parking_lot = { workspace = true }
 log = { workspace = true }
 bytes = { workspace = true, features = ["serde"] }
 sp-runtime = { workspace = true }
+fflonk = { git = "https://github.com/w3f/fflonk", features = ["std"]}
 
 [dev-dependencies]
 tonic = "0.10.2"
diff --git a/zk-gadget/src/client_ext.rs b/zk-gadget/src/client_ext.rs
index 83e4fde8a..7a69d7959 100644
--- a/zk-gadget/src/client_ext.rs
+++ b/zk-gadget/src/client_ext.rs
@@ -9,14 +9,14 @@ pub trait ClientWithApi<B: Block>: Client<B> + Clone {
     async fn get_job_circuit_properties(
         &self,
         circuit_id: u64,
-    ) -> Result<Option<CircuitProperties>, webb_gadget::Error>;
+    ) -> Result<Option<CircuitProperties>, gadget_common::Error>;
 
     async fn get_job_properties(
         &self,
         job_id: u64,
-    ) -> Result<Option<JobProperties>, webb_gadget::Error>;
+    ) -> Result<Option<JobProperties>, gadget_common::Error>;
 
-    async fn get_next_job(&self) -> Result<Option<JobProperties>, webb_gadget::Error>;
+    async fn get_next_job(&self) -> Result<Option<JobProperties>, gadget_common::Error>;
 }
 
 pub mod job_types {
diff --git a/zk-gadget/src/lib.rs b/zk-gadget/src/lib.rs
index d9db86b23..3d1714249 100644
--- a/zk-gadget/src/lib.rs
+++ b/zk-gadget/src/lib.rs
@@ -2,11 +2,11 @@ use crate::client_ext::ClientWithApi;
 use crate::network::{RegistantId, ZkNetworkService};
 use crate::protocol::proto_gen::{AsyncProtocolGenerator, ZkProtocolGenerator};
 use crate::protocol::{AdditionalProtocolParams, ZkProtocol};
+use gadget_common::Error;
 use mpc_net::prod::RustlsCertificate;
 use sp_runtime::traits::Block;
 use std::net::SocketAddr;
 use tokio_rustls::rustls::{Certificate, PrivateKey, RootCertStore};
-use webb_gadget::Error;
 
 pub mod network;
 pub mod protocol;
@@ -50,13 +50,14 @@ pub async fn run<
 
     let zk_protocol = ZkProtocol {
         party_id: config.id,
+        additional_params: (),
         client: client.clone(),
         protocol,
         _pd: std::marker::PhantomData,
     };
 
     // Plug the module into the webb gadget
-    webb_gadget::run_protocol(network, zk_protocol, client).await
+    gadget_common::run_protocol(network, zk_protocol, client).await
 }
 
 async fn create_zk_network(config: &ZkGadgetConfig) -> Result<ZkNetworkService, Error> {
diff --git a/zk-gadget/src/network/mod.rs b/zk-gadget/src/network/mod.rs
index 8b1d53d4e..8e6a65f18 100644
--- a/zk-gadget/src/network/mod.rs
+++ b/zk-gadget/src/network/mod.rs
@@ -49,9 +49,9 @@ pub struct Registrant {
 }
 
 use crate::Error;
-use webb_gadget::gadget::message::{GadgetProtocolMessage, UserID};
-use webb_gadget::gadget::network::Network;
-use webb_gadget::gadget::work_manager::WebbWorkManager;
+use gadget_common::gadget::message::{GadgetProtocolMessage, UserID};
+use gadget_common::gadget::network::Network;
+use gadget_common::gadget::work_manager::WebbWorkManager;
 
 pub fn create_server_tls_acceptor<T: CertToDer>(
     server_certificate: T,
diff --git a/zk-gadget/src/protocol/mod.rs b/zk-gadget/src/protocol/mod.rs
index 13957c0bb..cd90aee84 100644
--- a/zk-gadget/src/protocol/mod.rs
+++ b/zk-gadget/src/protocol/mod.rs
@@ -1,18 +1,19 @@
 use crate::client_ext::ClientWithApi;
 use crate::network::RegistantId;
 use async_trait::async_trait;
+use gadget_common::gadget::work_manager::WebbWorkManager;
+use gadget_common::gadget::{Job, WebbGadgetProtocol};
+use gadget_common::protocol::AsyncProtocol;
+use gadget_common::{BlockImportNotification, Error, FinalityNotification};
 use gadget_core::job_manager::ProtocolWorkManager;
 use sp_runtime::traits::Block;
-use webb_gadget::gadget::work_manager::WebbWorkManager;
-use webb_gadget::gadget::{Job, WebbGadgetProtocol};
-use webb_gadget::protocol::AsyncProtocol;
-use webb_gadget::{BlockImportNotification, Error, FinalityNotification};
 
 pub mod proto_gen;
 
-pub struct ZkProtocol<M, B, C> {
+pub struct ZkProtocol<M: AsyncProtocol, B, C> {
     pub party_id: RegistantId,
     pub protocol: M,
+    pub additional_params: <M as AsyncProtocol>::AdditionalParams,
     pub client: C,
     pub _pd: std::marker::PhantomData<B>,
 }
@@ -23,13 +24,15 @@ impl<T: Send + Sync + Clone + 'static> AdditionalProtocolParams for T {}
 #[async_trait]
 impl<M: AsyncProtocol + Send + Sync, B: Block, C: ClientWithApi<B>> WebbGadgetProtocol<B>
     for ZkProtocol<M, B, C>
+where
+    <M as AsyncProtocol>::AdditionalParams: Clone,
 {
-    async fn get_next_job(
+    async fn get_next_jobs(
         &self,
         _notification: &FinalityNotification<B>,
         now: u64,
         job_manager: &ProtocolWorkManager<WebbWorkManager>,
-    ) -> Result<Option<Job>, Error> {
+    ) -> Result<Option<Vec<Job>>, Error> {
         log::info!(
             "Party {} received a finality notification at {now}",
             self.party_id
@@ -46,11 +49,11 @@ impl<M: AsyncProtocol + Send + Sync, B: Block, C: ClientWithApi<B>> WebbGadgetPr
 
             let (remote, protocol) = self
                 .protocol
-                .create(now / 6000, now, 0, task_id)
+                .create(now / 6000, now, 0, task_id, self.additional_params.clone())
                 .await
                 .map_err(|err| Error::ClientError { err: err.reason })?;
 
-            return Ok(Some((remote, protocol)));
+            return Ok(Some(vec![(remote, protocol)]));
         } else {
             log::debug!("No Jobs found at #{now}");
         }
diff --git a/zk-gadget/src/protocol/proto_gen.rs b/zk-gadget/src/protocol/proto_gen.rs
index f73ad11ee..f23772e92 100644
--- a/zk-gadget/src/protocol/proto_gen.rs
+++ b/zk-gadget/src/protocol/proto_gen.rs
@@ -3,6 +3,10 @@ use crate::network::RegistantId;
 use crate::protocol::AdditionalProtocolParams;
 use async_trait::async_trait;
 use bytes::Bytes;
+use gadget_common::gadget::message::GadgetProtocolMessage;
+use gadget_common::gadget::network::Network;
+use gadget_common::gadget::work_manager::WebbWorkManager;
+use gadget_common::protocol::AsyncProtocol;
 use gadget_core::job::{BuiltExecutableJobWrapper, JobError};
 use gadget_core::job_manager::WorkManagerInterface;
 use mpc_net::{MpcNet, MpcNetError, MultiplexedStreamID};
@@ -12,14 +16,10 @@ use std::collections::HashMap;
 use std::fmt::Debug;
 use std::marker::PhantomData;
 use tokio::sync::mpsc::UnboundedReceiver;
-use webb_gadget::gadget::message::GadgetProtocolMessage;
-use webb_gadget::gadget::network::Network;
-use webb_gadget::gadget::work_manager::WebbWorkManager;
-use webb_gadget::protocol::AsyncProtocol;
 
 pub struct ZkAsyncProtocolParameters<B, N, C, Bl> {
     pub associated_block_id: <WebbWorkManager as WorkManagerInterface>::Clock,
-    pub associated_ssid: <WebbWorkManager as WorkManagerInterface>::SSID,
+    pub associated_retry_id: <WebbWorkManager as WorkManagerInterface>::RetryID,
     pub associated_session_id: <WebbWorkManager as WorkManagerInterface>::SessionID,
     pub associated_task_id: <WebbWorkManager as WorkManagerInterface>::TaskID,
     rxs: HashMap<u32, Vec<tokio::sync::Mutex<UnboundedReceiver<MpcNetMessage>>>>,
@@ -69,13 +69,15 @@ pub struct ZkProtocolGenerator<B, E, N, C, Bl> {
 impl<B: AdditionalProtocolParams, E, N: Network, C: ClientWithApi<Bl>, Bl: Block> AsyncProtocol
     for ZkProtocolGenerator<B, E, N, C, Bl>
 {
+    type AdditionalParams = ();
     async fn generate_protocol_from(
         &self,
         associated_block_id: <WebbWorkManager as WorkManagerInterface>::Clock,
-        associated_ssid: <WebbWorkManager as WorkManagerInterface>::SSID,
+        associated_retry_id: <WebbWorkManager as WorkManagerInterface>::RetryID,
         associated_session_id: <WebbWorkManager as WorkManagerInterface>::SessionID,
         associated_task_id: <WebbWorkManager as WorkManagerInterface>::TaskID,
         mut protocol_message_rx: UnboundedReceiver<GadgetProtocolMessage>,
+        _additional_params: Self::AdditionalParams,
     ) -> Result<BuiltExecutableJobWrapper, JobError> {
         let mut txs = HashMap::new();
         let mut rxs = HashMap::new();
@@ -128,7 +130,7 @@ impl<B: AdditionalProtocolParams, E, N: Network, C: ClientWithApi<Bl>, Bl: Block
 
         let params = ZkAsyncProtocolParameters {
             associated_block_id,
-            associated_ssid,
+            associated_retry_id,
             associated_session_id,
             associated_task_id,
             rxs,
@@ -199,11 +201,13 @@ impl<B: Send + Sync, N: Network, C: ClientWithApi<Bl>, Bl: Block> MpcNet
             .send_message(GadgetProtocolMessage {
                 associated_block_id: self.associated_block_id,
                 associated_session_id: self.associated_session_id,
-                associated_ssid: self.associated_ssid,
+                associated_retry_id: self.associated_retry_id,
                 from: self.party_id,
                 to: Some(id),
                 task_hash: self.associated_task_id,
                 payload: bincode2::serialize(&inner_payload).expect("Failed to serialize message"),
+                from_account_id: None,
+                to_account_id: None,
             })
             .await
             .map_err(|err| MpcNetError::Protocol {


Test patch:
diff --git a/test-gadget/Cargo.toml b/test-gadget/Cargo.toml
index a200d246e..af61fdd9e 100644
--- a/test-gadget/Cargo.toml
+++ b/test-gadget/Cargo.toml
@@ -24,11 +24,11 @@ mpc-net = { workspace = true }
 dist-primitives = { workspace = true }
 secret-sharing = { workspace = true }
 bytes = { version = "1.5.0", features = ["serde"]}
-ark-std = {version = "0.4.0", default-features = false, features = [ "print-trace", "std" ]}
-ark-ec = {version = "0.4.0", default-features = false}
-ark-bls12-377 = { version = "0.4.0", default-features = false, features = ["curve"] }
-ark-ff = {version = "0.4.0", default-features = false}
-ark-poly = {version = "0.4.0", default-features = false}
+ark-std = { version = "*", features = [ "print-trace", "std" ]}
+ark-ec = { version = "*"}
+ark-bls12-377 = { version = "*", features = ["curve"] }
+ark-ff = { version = "*"}
+ark-poly = { version = "*"}
 bincode2 = "2.0.1"
 hex = "0.4.3"
 md5 = "0.7.0"
diff --git a/test-gadget/src/gadget.rs b/test-gadget/src/gadget.rs
index 95fbbc124..c0076b61a 100644
--- a/test-gadget/src/gadget.rs
+++ b/test-gadget/src/gadget.rs
@@ -90,11 +90,11 @@ impl<B: Send + Sync + Clone + 'static> AbstractGadget for TestGadget<B> {
         if self.run_test_at.contains(&now) {
             log::info!("Running test at block {now}");
             let task_hash = now.to_be_bytes();
-            let ssid = 0; // Assume SSID = 0 for now
+            let retry_id = 0; // Assume SSID = 0 for now
             let (remote, task) = create_test_async_protocol(
                 session_id,
                 now,
-                ssid,
+                retry_id,
                 task_hash,
                 self.test_bundle.clone(),
                 &*self.async_protocol_generator,
@@ -137,7 +137,7 @@ impl<B: Send + Sync + Clone + 'static> AbstractGadget for TestGadget<B> {
 fn create_test_async_protocol<B: Send + Sync + 'static>(
     session_id: <TestWorkManager as WorkManagerInterface>::SessionID,
     now: <TestWorkManager as WorkManagerInterface>::Clock,
-    ssid: <TestWorkManager as WorkManagerInterface>::SSID,
+    retry_id: <TestWorkManager as WorkManagerInterface>::RetryID,
     task_id: <TestWorkManager as WorkManagerInterface>::TaskID,
     test_bundle: B,
     proto_gen: &dyn AsyncProtocolGenerator<B>,
@@ -153,7 +153,7 @@ fn create_test_async_protocol<B: Send + Sync + 'static>(
         start_rx: Some(start_rx),
         shutdown_rx: Some(shutdown_rx),
         associated_block_id: now,
-        associated_ssid: ssid,
+        associated_retry_id: retry_id,
         associated_session_id: session_id,
         associated_task_id: task_id,
         test_bundle,
@@ -163,14 +163,14 @@ fn create_test_async_protocol<B: Send + Sync + 'static>(
         start_tx: parking_lot::Mutex::new(Some(start_tx)),
         shutdown_tx: parking_lot::Mutex::new(Some(shutdown_tx)),
         associated_block_id: now,
-        associated_ssid: ssid,
+        associated_retry_id: retry_id,
         associated_session_id: session_id,
         to_async_protocol,
         is_done,
     };
 
     let future = proto_gen(params);
-    let future = JobBuilder::default().build(future);
+    let future = JobBuilder::default().protocol(future).build();
 
     (remote, future)
 }
diff --git a/test-gadget/src/message.rs b/test-gadget/src/message.rs
index bca454d5a..2eeb42405 100644
--- a/test-gadget/src/message.rs
+++ b/test-gadget/src/message.rs
@@ -9,7 +9,7 @@ pub struct TestProtocolMessage {
     pub to: Option<UserID>,
     pub associated_block_id: <TestWorkManager as WorkManagerInterface>::Clock,
     pub associated_session_id: <TestWorkManager as WorkManagerInterface>::SessionID,
-    pub associated_ssid: <TestWorkManager as WorkManagerInterface>::SSID,
+    pub associated_retry_id: <TestWorkManager as WorkManagerInterface>::RetryID,
     pub associated_task_id: <TestWorkManager as WorkManagerInterface>::TaskID,
 }
 
@@ -24,11 +24,21 @@ impl ProtocolMessageMetadata<TestWorkManager> for TestProtocolMessage {
         self.associated_session_id
     }
 
-    fn associated_ssid(&self) -> <TestWorkManager as WorkManagerInterface>::SSID {
-        self.associated_ssid
+    fn associated_retry_id(&self) -> <TestWorkManager as WorkManagerInterface>::RetryID {
+        self.associated_retry_id
     }
 
     fn associated_task(&self) -> <TestWorkManager as WorkManagerInterface>::TaskID {
         self.associated_task_id
     }
+
+    fn associated_sender_user_id(&self) -> <TestWorkManager as WorkManagerInterface>::UserID {
+        self.from
+    }
+
+    fn associated_recipient_user_id(
+        &self,
+    ) -> Option<<TestWorkManager as WorkManagerInterface>::UserID> {
+        self.to
+    }
 }
diff --git a/test-gadget/src/work_manager.rs b/test-gadget/src/work_manager.rs
index e48ac526a..a1f36dc46 100644
--- a/test-gadget/src/work_manager.rs
+++ b/test-gadget/src/work_manager.rs
@@ -1,5 +1,5 @@
 use crate::error::TestError;
-use crate::message::TestProtocolMessage;
+use crate::message::{TestProtocolMessage, UserID};
 use gadget_core::job::JobError;
 use gadget_core::job_manager::{ProtocolRemote, SendFuture, ShutdownReason, WorkManagerInterface};
 use parking_lot::{Mutex, RwLock};
@@ -16,7 +16,7 @@ pub struct TestProtocolRemote {
     pub shutdown_tx: Mutex<Option<tokio::sync::oneshot::Sender<ShutdownReason>>>,
     pub associated_session_id: <TestWorkManager as WorkManagerInterface>::SessionID,
     pub associated_block_id: <TestWorkManager as WorkManagerInterface>::Clock,
-    pub associated_ssid: <TestWorkManager as WorkManagerInterface>::SSID,
+    pub associated_retry_id: <TestWorkManager as WorkManagerInterface>::RetryID,
     pub to_async_protocol: tokio::sync::mpsc::UnboundedSender<
         <TestWorkManager as WorkManagerInterface>::ProtocolMessage,
     >,
@@ -26,8 +26,8 @@ pub struct TestProtocolRemote {
 const ACCEPTABLE_BLOCK_TOLERANCE: u64 = 20;
 
 impl WorkManagerInterface for TestWorkManager {
-    // TODO: rename to retry_id or similar
-    type SSID = u16;
+    type RetryID = u16;
+    type UserID = UserID;
     type Clock = u64;
     type ProtocolMessage = TestProtocolMessage;
     type Error = TestError;
@@ -73,8 +73,6 @@ impl ProtocolRemote<TestWorkManager> for TestProtocolRemote {
         self.associated_session_id
     }
 
-    fn set_as_primary(&self) {}
-
     fn started_at(&self) -> <TestWorkManager as WorkManagerInterface>::Clock {
         self.associated_block_id
     }
@@ -112,8 +110,8 @@ impl ProtocolRemote<TestWorkManager> for TestProtocolRemote {
         self.start_tx.lock().is_none()
     }
 
-    fn ssid(&self) -> <TestWorkManager as WorkManagerInterface>::SSID {
-        self.associated_ssid
+    fn retry_id(&self) -> <TestWorkManager as WorkManagerInterface>::RetryID {
+        self.associated_retry_id
     }
 }
 
@@ -125,7 +123,7 @@ pub struct TestAsyncProtocolParameters<B> {
     pub start_rx: Option<tokio::sync::oneshot::Receiver<()>>,
     pub shutdown_rx: Option<tokio::sync::oneshot::Receiver<ShutdownReason>>,
     pub associated_block_id: <TestWorkManager as WorkManagerInterface>::Clock,
-    pub associated_ssid: <TestWorkManager as WorkManagerInterface>::SSID,
+    pub associated_retry_id: <TestWorkManager as WorkManagerInterface>::RetryID,
     pub associated_session_id: <TestWorkManager as WorkManagerInterface>::SessionID,
     pub associated_task_id: <TestWorkManager as WorkManagerInterface>::TaskID,
     pub test_bundle: B,
diff --git a/test-gadget/tests/basic.rs b/test-gadget/tests/basic.rs
index 76789a1cb..e5d53b7c5 100644
--- a/test-gadget/tests/basic.rs
+++ b/test-gadget/tests/basic.rs
@@ -58,7 +58,7 @@ mod tests {
                         to: None,
                         associated_block_id: params.associated_block_id,
                         associated_session_id: params.associated_session_id,
-                        associated_ssid: params.associated_ssid,
+                        associated_retry_id: params.associated_retry_id,
                         associated_task_id: params.associated_task_id,
                     },
                 )
diff --git a/test-gadget/tests/dfft.rs b/test-gadget/tests/dfft.rs
index e45d477c3..eca0ae60d 100644
--- a/test-gadget/tests/dfft.rs
+++ b/test-gadget/tests/dfft.rs
@@ -61,19 +61,9 @@ mod tests {
             .collect::<Vec<_>>();
 
         // Rearranging x
-
-        let peval_share = d_fft(
-            pcoeff_share,
-            false,
-            1,
-            false,
-            dom,
-            pp,
-            net,
-            MultiplexedStreamID::One,
-        )
-        .await
-        .unwrap();
+        let peval_share = d_fft(pcoeff_share, false, dom, pp, net, MultiplexedStreamID::One)
+            .await
+            .unwrap();
 
         // Send to king who reconstructs and checks the answer
         let result = net
@@ -159,7 +149,7 @@ mod tests {
                 party_id: params.test_bundle.party_id,
                 associated_block_id: params.associated_block_id,
                 associated_session_id: params.associated_session_id,
-                associated_ssid: params.associated_ssid,
+                associated_retry_id: params.associated_retry_id,
                 associated_task_id: params.associated_task_id,
             };
 
@@ -198,7 +188,7 @@ mod tests {
         party_id: u32,
         associated_block_id: u64,
         associated_session_id: u64,
-        associated_ssid: u16,
+        associated_retry_id: u16,
         associated_task_id: [u8; 8],
     }
 
@@ -265,7 +255,7 @@ mod tests {
             to: Some(to),
             associated_block_id: network.associated_block_id,
             associated_session_id: network.associated_session_id,
-            associated_ssid: network.associated_ssid,
+            associated_retry_id: network.associated_retry_id,
             associated_task_id: network.associated_task_id,
         };
 
diff --git a/test-gadget/tests/dmsm.rs b/test-gadget/tests/dmsm.rs
index b13a9ce3d..0ef574681 100644
--- a/test-gadget/tests/dmsm.rs
+++ b/test-gadget/tests/dmsm.rs
@@ -154,7 +154,7 @@ mod tests {
                 party_id: params.test_bundle.party_id,
                 associated_block_id: params.associated_block_id,
                 associated_session_id: params.associated_session_id,
-                associated_ssid: params.associated_ssid,
+                associated_retry_id: params.associated_retry_id,
                 associated_task_id: params.associated_task_id,
             };
 
@@ -183,7 +183,7 @@ mod tests {
         party_id: u32,
         associated_block_id: u64,
         associated_session_id: u64,
-        associated_ssid: u16,
+        associated_retry_id: u16,
         associated_task_id: [u8; 8],
     }
 
@@ -244,7 +244,7 @@ mod tests {
             to,
             associated_block_id: network.associated_block_id,
             associated_session_id: network.associated_session_id,
-            associated_ssid: network.associated_ssid,
+            associated_retry_id: network.associated_retry_id,
             associated_task_id: network.associated_task_id,
         };
 
diff --git a/test-gadget/tests/zk_network.rs b/test-gadget/tests/zk_network.rs
index 0a1fc7564..9d926c430 100644
--- a/test-gadget/tests/zk_network.rs
+++ b/test-gadget/tests/zk_network.rs
@@ -79,7 +79,7 @@ mod tests {
                 party_id: params.test_bundle.party_id,
                 associated_block_id: params.associated_block_id,
                 associated_session_id: params.associated_session_id,
-                associated_ssid: params.associated_ssid,
+                associated_retry_id: params.associated_retry_id,
                 associated_task_id: params.associated_task_id,
             };
 
@@ -157,7 +157,7 @@ mod tests {
         party_id: u32,
         associated_block_id: u64,
         associated_session_id: u64,
-        associated_ssid: u16,
+        associated_retry_id: u16,
         associated_task_id: [u8; 8],
     }
 
@@ -218,7 +218,7 @@ mod tests {
             to,
             associated_block_id: network.associated_block_id,
             associated_session_id: network.associated_session_id,
-            associated_ssid: network.associated_ssid,
+            associated_retry_id: network.associated_retry_id,
             associated_task_id: network.associated_task_id,
         };
 
diff --git a/zk-gadget/tests/jobs.rs b/zk-gadget/tests/jobs.rs
index f77cb3108..79ef6966a 100644
--- a/zk-gadget/tests/jobs.rs
+++ b/zk-gadget/tests/jobs.rs
@@ -56,6 +56,7 @@ mod tests {
 
     pub mod client {
         use async_trait::async_trait;
+        use gadget_common::{BlockImportNotification, FinalityNotification};
         use gadget_core::gadget::substrate::Client;
         use sc_client_api::FinalizeSummary;
         use serde::Serialize;
@@ -67,7 +68,6 @@ mod tests {
         use std::time::Duration;
         use tokio::sync::Mutex;
         use uuid::Uuid;
-        use webb_gadget::{BlockImportNotification, FinalityNotification};
         use zk_gadget::client_ext::job_types::{CircuitProperties, JobProperties};
         use zk_gadget::client_ext::ClientWithApi;
 
@@ -137,18 +137,18 @@ mod tests {
             async fn get_job_circuit_properties(
                 &self,
                 _circuit_id: u64,
-            ) -> Result<Option<CircuitProperties>, webb_gadget::Error> {
+            ) -> Result<Option<CircuitProperties>, gadget_common::Error> {
                 unimplemented!()
             }
 
             async fn get_job_properties(
                 &self,
                 _job_id: u64,
-            ) -> Result<Option<JobProperties>, webb_gadget::Error> {
+            ) -> Result<Option<JobProperties>, gadget_common::Error> {
                 unimplemented!()
             }
 
-            async fn get_next_job(&self) -> Result<Option<JobProperties>, webb_gadget::Error> {
+            async fn get_next_job(&self) -> Result<Option<JobProperties>, gadget_common::Error> {
                 Ok(Some(JobProperties {
                     job_id: 0,
                     circuit_id: 0,
@@ -335,45 +335,47 @@ mod tests {
             TestBlock,
         >,
     ) -> BuiltExecutableJobWrapper {
-        JobBuilder::default().build(async move {
-            if params.party_id == 0 {
-                // Receive N-1 messages from the other parties
-                for party_id in 0..N {
-                    let party_id = party_id as u32;
-                    if party_id != params.party_id {
-                        let _message = params
-                            .recv_from(party_id, MultiplexedStreamID::Zero)
-                            .await
-                            .expect("Should receive protocol message");
+        JobBuilder::default()
+            .protocol(async move {
+                if params.party_id == 0 {
+                    // Receive N-1 messages from the other parties
+                    for party_id in 0..N {
+                        let party_id = party_id as u32;
+                        if party_id != params.party_id {
+                            let _message = params
+                                .recv_from(party_id, MultiplexedStreamID::Zero)
+                                .await
+                                .expect("Should receive protocol message");
+                        }
                     }
-                }
 
-                for party_id in 0..N {
-                    let party_id = party_id as u32;
-                    if party_id != params.party_id {
-                        params
-                            .send_to(party_id, Default::default(), MultiplexedStreamID::Zero)
-                            .await
-                            .expect("Should send");
+                    for party_id in 0..N {
+                        let party_id = party_id as u32;
+                        if party_id != params.party_id {
+                            params
+                                .send_to(party_id, Default::default(), MultiplexedStreamID::Zero)
+                                .await
+                                .expect("Should send");
+                        }
                     }
+                } else {
+                    params
+                        .send_to(0, Default::default(), MultiplexedStreamID::Zero)
+                        .await
+                        .expect("Should send");
+                    let _message_from_king = params
+                        .recv_from(0, MultiplexedStreamID::Zero)
+                        .await
+                        .expect("Should receive protocol message");
                 }
-            } else {
+
                 params
-                    .send_to(0, Default::default(), MultiplexedStreamID::Zero)
-                    .await
+                    .extra_parameters
+                    .stop_tx
+                    .send(())
                     .expect("Should send");
-                let _message_from_king = params
-                    .recv_from(0, MultiplexedStreamID::Zero)
-                    .await
-                    .expect("Should receive protocol message");
-            }
-
-            params
-                .extra_parameters
-                .stop_tx
-                .send(())
-                .expect("Should send");
-            Ok(())
-        })
+                Ok(())
+            })
+            .build()
     }
 }