Merge branch 'master' of github.com:darkrenaissance/darkfi

2026-04-28 03:00:18 -04:00 · 2021-06-09 10:25:57 +02:00
parent 1ff8cfa793 40e79b2c1d
commit de47e7c4e1
22 changed files with 384 additions and 199 deletions
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -18,9 +18,9 @@ checksum = "d9fe5e32de01730eb1f6b7f5b51c17e03e2325bf40a74f754f04f130043affff"

 [[package]]
 name = "addr2line"
-version = "0.15.1"
+version = "0.15.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "03345e98af8f3d786b6d9f656ccfa6ac316d954e92bc4841f0bba20789d5fb5a"
+checksum = "e7a2e47a1fbe209ee101dd6d61285226744c6c8d3c21c8dc878ba6cb9f467f3a"
 dependencies = [
 "gimli",
 ]
@@ -156,9 +156,9 @@ dependencies = [

 [[package]]
 name = "approx"
-version = "0.4.0"
+version = "0.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3f2a05fd1bd10b2527e20a2cd32d8873d115b8b39fe219ee25f42a8aca6ba278"
+checksum = "072df7202e63b127ab55acfe16ce97013d5b97bf160489336d3f1840fd78e99e"
 dependencies = [
 "num-traits",
 ]
@@ -452,9 +452,9 @@ checksum = "cdb031dd78e28731d87d56cc8ffef4a8f36ca26c38fe2de700543e627f8a464a"

 [[package]]
 name = "backtrace"
-version = "0.3.59"
+version = "0.3.60"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4717cfcbfaa661a0fd48f8453951837ae7e8f81e481fbb136e3202d72805a744"
+checksum = "b7815ea54e4d821e791162e078acbebfd6d8c8939cd559c9335dceb1c8ca7282"
 dependencies = [
 "addr2line",
 "cc",
@@ -1715,9 +1715,9 @@ checksum = "acc499defb3b348f8d8f3f66415835a9131856ff7714bf10dadfc4ec4bdb29a1"

 [[package]]
 name = "futures-lite"
-version = "1.11.3"
+version = "1.12.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b4481d0cd0de1d204a4fa55e7d45f07b1d958abcb06714b3446438e2eff695fb"
+checksum = "7694489acd39452c77daa48516b894c153f192c3578d5a839b62c58099fcbf48"
 dependencies = [
 "fastrand",
 "futures-core",
@@ -2049,12 +2049,12 @@ dependencies = [

 [[package]]
 name = "glyph_brush_layout"
-version = "0.2.1"
+version = "0.2.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "10bc06d530bf20c1902f1b02799ab7372ff43f6119770c49b0bc3f21bd148820"
+checksum = "15cf18cf985bd942f05e14552b63c9d08f7d0ed1ec79a977eb9747c9e065f497"
 dependencies = [
 "ab_glyph",
- "approx 0.4.0",
+ "approx 0.5.0",
 "xi-unicode",
 ]

@@ -2140,9 +2140,9 @@ dependencies = [

 [[package]]
 name = "heck"
-version = "0.3.2"
+version = "0.3.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "87cbf45460356b7deeb5e3415b5563308c0a9b057c85e12b06ad551f98d0a6ac"
+checksum = "6d621efb26863f0e9924c6ac577e8275e5e6b77455db64ffa6c65c904e9e132c"
 dependencies = [
 "unicode-segmentation",
 ]
@@ -2885,9 +2885,12 @@ dependencies = [

 [[package]]
 name = "object"
-version = "0.24.0"
+version = "0.25.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1a5b3dd1c072ee7963717671d1ca129f1048fda25edea6b752bfc71ac8854170"
+checksum = "f8bc1d42047cf336f0f939c99e97183cf31551bf0f2865a2ec9c8d91fd4ffb5e"
+dependencies = [
+ "memchr",
+]

 [[package]]
 name = "once_cell"
@@ -3585,9 +3588,9 @@ checksum = "d29ab0c6d3fc0ee92fe66e2d99f700eab17a8d57d1c1d3b748380fb20baa78cd"

 [[package]]
 name = "security-framework"
-version = "2.2.0"
+version = "2.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3670b1d2fdf6084d192bc71ead7aabe6c06aa2ea3fbd9cc3ac111fa5c2b1bd84"
+checksum = "b239a3d5db51252f6f48f42172c65317f37202f4a21021bf5f9d40a408f4592c"
 dependencies = [
 "bitflags",
 "core-foundation 0.9.1",
@@ -3598,9 +3601,9 @@ dependencies = [

 [[package]]
 name = "security-framework-sys"
-version = "2.2.0"
+version = "2.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3676258fd3cfe2c9a0ec99ce3038798d847ce3e4bb17746373eb9f0f1ac16339"
+checksum = "7e4effb91b4b8b6fb7732e670b6cee160278ff8e6bf485c7805d9e319d76e284"
 dependencies = [
 "core-foundation-sys 0.8.2",
 "libc",
@@ -3723,9 +3726,9 @@ checksum = "7fdf1b9db47230893d76faad238fd6097fd6d6a9245cd7a4d90dbd639536bbd2"

 [[package]]
 name = "signal-hook"
-version = "0.3.8"
+version = "0.3.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ef33d6d0cd06e0840fba9985aab098c147e67e05cee14d412d3345ed14ff30ac"
+checksum = "470c5a6397076fae0094aaf06a08e6ba6f37acb77d3b1b91ea92b4d6c8650c39"
 dependencies = [
 "libc",
 "signal-hook-registry",
@@ -3733,9 +3736,9 @@ dependencies = [

 [[package]]
 name = "signal-hook-registry"
-version = "1.3.0"
+version = "1.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "16f1d0fef1604ba8f7a073c7e701f213e056707210e9020af4528e0101ce11a6"
+checksum = "e51e73328dc4ac0c7ccbda3a494dfa03df1de2f46018127f60c693f2648455b0"
 dependencies = [
 "libc",
 ]
@@ -4183,7 +4186,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "04f8ab788026715fa63b31960869617cba39117e520eb415b0139543e325ab59"
 dependencies = [
 "cfg-if 0.1.10",
- "rand 0.6.5",
+ "rand 0.7.3",
 "static_assertions",
 ]

@@ -4204,9 +4207,9 @@ dependencies = [

 [[package]]
 name = "unicode-normalization"
-version = "0.1.18"
+version = "0.1.19"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "33717dca7ac877f497014e10d73f3acf948c342bee31b5ca7892faf94ccc6b49"
+checksum = "d54590932941a9e9266f0832deed84ebe1bf2e4c9e4a3554d393d18f5e854bf9"
 dependencies = [
 "tinyvec",
 ]
--- a/lisp/core.rs
+++ b/lisp/core.rs
@@ -705,12 +705,12 @@ fn seq(a: MalArgs) -> MalRet {
    }
 }

-fn gen_rand(a: MalArgs) -> MalRet {
+fn gen_rand(_a: MalArgs) -> MalRet {
    let mut rng = rand::thread_rng();
    Ok(MalVal::Int(rng.gen::<i64>()))
 }

-fn scalar_rnd(a: MalArgs) -> MalRet {
+fn scalar_rnd(_a: MalArgs) -> MalRet {
    let randomness_value: jubjub::Fr = jubjub::Fr::random(&mut OsRng);
    let value = bls12_381::Scalar::from_bytes(&randomness_value.to_bytes());
    Ok(MalVal::ZKScalar(value.unwrap()))
--- a/lisp/lisp.rs
+++ b/lisp/lisp.rs
@@ -138,22 +138,22 @@ fn eval_ast(ast: &MalVal, env: &Env) -> MalRet {
    }
 }

-fn eval(mut ast: MalVal, mut env: Env) -> MalRet {
+fn eval(mut _ast: MalVal, mut env: Env) -> MalRet {
    let ret: MalRet;

    let start = Instant::now();

    'tco: loop {
        // TODO check DEBUG symbol on env
-        println!("debug eval \t {:?} \t {:?}", ast, start.elapsed());
-        ret = match ast.clone() {
+        println!("debug eval \t {:?} \t {:?}", _ast, start.elapsed());
+        ret = match _ast.clone() {
            List(l, _) => {
                if l.len() == 0 {
-                    return Ok(ast);
+                    return Ok(_ast);
                }
-                match macroexpand(ast.clone(), &env) {
+                match macroexpand(_ast.clone(), &env) {
                    (true, Ok(new_ast)) => {
-                        ast = new_ast;
+                        _ast = new_ast;
                        continue 'tco;
                    }
                    (_, Err(e)) => return Err(e),
@@ -161,7 +161,7 @@ fn eval(mut ast: MalVal, mut env: Env) -> MalRet {
                }

                if l.len() == 0 {
-                    return Ok(ast);
+                    return Ok(_ast);
                }
                let a0 = &l[0];
                match a0 {
@@ -192,7 +192,7 @@ fn eval(mut ast: MalVal, mut env: Env) -> MalRet {
                                return error("let* with non-List bindings");
                            }
                        };
-                        ast = a2;
+                        _ast = a2;
                        continue 'tco;
                    }
                    Sym(ref a0sym) if a0sym == "let*" => {
@@ -219,13 +219,13 @@ fn eval(mut ast: MalVal, mut env: Env) -> MalRet {
                                return error("let* with non-List bindings");
                            }
                        };
-                        ast = a2;
+                        _ast = a2;
                        continue 'tco;
                    }
                    Sym(ref a0sym) if a0sym == "quote" => Ok(l[1].clone()),
                    Sym(ref a0sym) if a0sym == "quasiquoteexpand" => Ok(quasiquote(&l[1])),
                    Sym(ref a0sym) if a0sym == "quasiquote" => {
-                        ast = quasiquote(&l[1]);
+                        _ast = quasiquote(&l[1]);
                        continue 'tco;
                    }
                    Sym(ref a0sym) if a0sym == "defmacro!" => {
@@ -242,7 +242,7 @@ fn eval(mut ast: MalVal, mut env: Env) -> MalRet {
                                &env,
                                a1.clone(),
                                MalFunc {
-                                    eval: eval,
+                                    eval,
                                    ast: ast.clone(),
                                    env: env.clone(),
                                    params: params.clone(),
@@ -282,7 +282,7 @@ fn eval(mut ast: MalVal, mut env: Env) -> MalRet {
                    Sym(ref a0sym) if a0sym == "do" => {
                        match eval_ast(&list!(l[1..l.len() - 1].to_vec()), &env)? {
                            List(_, _) => {
-                                ast = l.last().unwrap_or(&Nil).clone();
+                                _ast = l.last().unwrap_or(&Nil).clone();
                                continue 'tco;
                            }
                            _ => error("invalid do form"),
@@ -292,7 +292,7 @@ fn eval(mut ast: MalVal, mut env: Env) -> MalRet {
                        match eval(l[1].clone(), env.clone())? {
                            MalVal::Int(v) => {
                                for _i in 0..v {
-                                    ast = eval_ast(&l[2], &env)?;
+                                    _ast = eval_ast(&l[2], &env)?;
                                }
                                Ok(Nil)
                            }
@@ -303,12 +303,12 @@ fn eval(mut ast: MalVal, mut env: Env) -> MalRet {
                        let cond = eval(l[1].clone(), env.clone())?;
                        match cond {
                            Bool(false) | Nil if l.len() >= 4 => {
-                                ast = l[3].clone();
+                                _ast = l[3].clone();
                                continue 'tco;
                            }
                            Bool(false) | Nil => Ok(Nil),
                            _ if l.len() >= 3 => {
-                                ast = l[2].clone();
+                                _ast = l[2].clone();
                                continue 'tco;
                            }
                            _ => Ok(Nil),
@@ -318,16 +318,16 @@ fn eval(mut ast: MalVal, mut env: Env) -> MalRet {
                    Sym(ref a0sym) if a0sym == "fn*" => {
                        let (a1, a2) = (l[1].clone(), l[2].clone());
                        Ok(MalFunc {
-                            eval: eval,
+                            eval,
                            ast: Rc::new(a2),
-                            env: env,
+                            env,
                            params: Rc::new(a1),
                            is_macro: false,
                            meta: Rc::new(Nil),
                        })
                    }
                    Sym(ref a0sym) if a0sym == "eval" => {
-                        ast = eval(l[1].clone(), env.clone())?;
+                        _ast = eval(l[1].clone(), env.clone())?;
                        while let Some(ref e) = env.clone().outer {
                            env = e.clone();
                        }
@@ -336,7 +336,7 @@ fn eval(mut ast: MalVal, mut env: Env) -> MalRet {
                    Sym(ref a0sym) if a0sym == "setup" => {
                        let a1 = l[1].clone();
                        // todo
-                        ast = eval(a1.clone(), env.clone())?;
+                        _ast = eval(a1.clone(), env.clone())?;
                        //                        let _pvk = setup(a1.clone(), env.clone())?;
                        continue 'tco;
                    }
@@ -347,7 +347,7 @@ fn eval(mut ast: MalVal, mut env: Env) -> MalRet {
                        prove(a1.clone(), env.clone())
                    }
                    Sym(ref a0sym) if a0sym == "kill" => {
-                        error(&format!("KILL at: {:?}", ast).to_string())
+                        error(&format!("KILL at: {:?}", _ast).to_string())
                    }
                    Sym(ref a0sym) if a0sym == "alloc-const" => {
                        let start = Instant::now();
@@ -536,7 +536,7 @@ fn eval(mut ast: MalVal, mut env: Env) -> MalRet {

                        Ok(MalVal::Str("enforce-eof".to_string()))
                    }
-                    _ => match eval_ast(&ast, &env)? {
+                    _ => match eval_ast(&_ast, &env)? {
                        List(ref el, _) => {
                            let ref f = el[0].clone();
                            let args = el[1..].to_vec();
@@ -551,7 +551,7 @@ fn eval(mut ast: MalVal, mut env: Env) -> MalRet {
                                    let a = &**mast;
                                    let p = &**params;
                                    env = env_bind(Some(menv.clone()), p.clone(), args)?;
-                                    ast = a.clone();
+                                    _ast = a.clone();
                                    continue 'tco;
                                }
                                _ => {
@@ -565,7 +565,7 @@ fn eval(mut ast: MalVal, mut env: Env) -> MalRet {
                    },
                }
            }
-            _ => eval_ast(&ast, &env),
+            _ => eval_ast(&_ast, &env),
        };

        break;
@@ -628,7 +628,7 @@ pub fn get_allocations_nested(env: &Env, key: &str) -> RefCell<HashMap<String, M
    }
 }

-pub fn setup(_ast: MalVal, env: Env) -> Result<VerifyKeyParams, MalErr> {
+pub fn setup(_ast: MalVal, _env: Env) -> Result<VerifyKeyParams, MalErr> {
    let start = Instant::now();
    let c = LispCircuit {
        params: HashMap::default(),
@@ -674,7 +674,7 @@ pub fn prove(_ast: MalVal, env: Env) -> MalRet {
    };
    let proof = groth16::create_random_proof(circuit, params.as_ref().unwrap(), &mut OsRng)?;
    let mut vec_input = vec![];
-    for (k, val) in allocs_input.borrow_mut().iter() {
+    for (_k, val) in allocs_input.borrow_mut().iter() {
        match val {
            MalVal::Str(v) => {
                vec_input.push(bls12_381::Scalar::from_string(&v.to_string()));
--- a/lisp/types.rs
+++ b/lisp/types.rs
@@ -124,7 +124,7 @@ impl Circuit<bls12_381::Scalar> for LispCircuit {
        }
        println!("circuit alloc input \t {:?}", start.elapsed());
        let start = Instant::now();
-        let mut enforce_sorted = self.constraints.clone();
+        let enforce_sorted = self.constraints.clone();
        // enforce_sorted.sort_by(|a, b| a.idx.cmp(&b.idx));
        for alloc_value in enforce_sorted.iter() {
            // println!("Enforce -> {:?}", alloc_value);
--- a/src/bin/darkfid.rs
+++ b/src/bin/darkfid.rs
@@ -2,7 +2,7 @@ use async_std::sync::Arc;
 use rand::rngs::OsRng;
 use std::net::SocketAddr;

-use drk::blockchain::{rocks::columns, Rocks, RocksColumn, Slab, SlabStore};
+use drk::blockchain::{rocks::columns, Rocks, RocksColumn};
 use drk::crypto::{
    coin::Coin,
    load_params,
@@ -13,11 +13,11 @@ use drk::crypto::{
    save_params, setup_mint_prover, setup_spend_prover,
 };
 use drk::serial::Decodable;
-use drk::service::{ClientProgramOptions, GatewayClient, Subscriber};
+use drk::service::{ClientProgramOptions, GatewayClient, GatewaySlabsSubscriber};
 use drk::state::{state_transition, ProgramState, StateUpdate};
 use drk::wallet::WalletDB;
 use drk::{tx, Result};
-use rusqlite::{named_params, Connection};
+use rusqlite::Connection;

 use async_executor::Executor;
 use bellman::groth16;
@@ -26,6 +26,7 @@ use easy_parallel::Parallel;
 use ff::Field;
 use std::path::Path;

+#[allow(dead_code)]
 pub struct State {
    // The entire merkle tree state
    tree: CommitmentTree<MerkleNode>,
@@ -153,19 +154,13 @@ fn setup_addr(address: Option<SocketAddr>, default: SocketAddr) -> SocketAddr {
    }
 }

-pub async fn subscribe(
-    mut subscriber: Subscriber,
-    slabstore: Arc<SlabStore>,
-    mut state: State,
-) -> Result<()> {
+pub async fn subscribe(gateway_slabs_sub: GatewaySlabsSubscriber, mut state: State) -> Result<()> {
    loop {
-        let slab = subscriber.fetch::<Slab>().await?;
+        let slab = gateway_slabs_sub.recv().await?;
        let tx = tx::Transaction::decode(&slab.get_payload()[..])?;

        let update = state_transition(&state, tx)?;
        state.apply(update)?;
-
-        slabstore.put(slab)?;
    }
 }

@@ -177,11 +172,6 @@ async fn start(executor: Arc<Executor<'_>>, options: ClientProgramOptions) -> Re
    let rocks = Rocks::new(database_path)?;

    let slabstore = RocksColumn::<columns::Slabs>::new(rocks.clone());
-    // create gateway client
-    let mut client = GatewayClient::new(connect_addr, slabstore)?;
-
-    // start gateway client
-    client.start().await?;

    //
    // Auto create trusted ceremony parameters if they don't exist
@@ -221,10 +211,16 @@ async fn start(executor: Arc<Executor<'_>>, options: ClientProgramOptions) -> Re
        secrets: vec![secret.clone()],
    };

-    // start subscribe to gateway publisher
-    let subscriber = GatewayClient::start_subscriber(sub_addr).await?;
-    let slabstore = client.get_slabstore();
-    let subscribe_task = executor.spawn(subscribe(subscriber, slabstore, state));
+    // create gateway client
+    let mut client = GatewayClient::new(connect_addr, slabstore)?;
+
+    // start subscribing
+    let gateway_slabs_sub: GatewaySlabsSubscriber =
+        client.start_subscriber(sub_addr, executor.clone()).await?;
+    let subscribe_task = executor.spawn(subscribe(gateway_slabs_sub, state));
+
+    // start gateway client
+    client.start().await?;

    subscribe_task.cancel().await;
    Ok(())
@@ -273,63 +269,199 @@ fn main() -> Result<()> {
    result
 }

-//
-//// $ cargo test --bin darkfid
-//// run 10 clients simultaneously
-//#[cfg(test)]
-//mod test {
-//
-//    #[test]
-//    fn test_darkfid_client() {
-//        use std::path::Path;
-//
-//        use drk::blockchain::{Rocks, Slab};
-//        use drk::service::GatewayClient;
-//
-//        use log::*;
-//        use rand::Rng;
-//        use simplelog::*;
-//
-//        let logger_config = ConfigBuilder::new().set_time_format_str("%T%.6f").build();
-//
-//        CombinedLogger::init(vec![
-//            TermLogger::new(LevelFilter::Debug, logger_config, TerminalMode::Mixed).unwrap(),
-//            WriteLogger::new(
-//                LevelFilter::Debug,
-//                Config::default(),
-//                std::fs::File::create(Path::new("/tmp/dar.log")).unwrap(),
-//            ),
-//        ])
-//        .unwrap();
-//
-//        let mut thread_pools: Vec<std::thread::JoinHandle<()>> = vec![];
-//
-//        for _ in 0..10 {
-//            let thread = std::thread::spawn(|| {
-//                smol::future::block_on(async move {
-//                    let mut rng = rand::thread_rng();
-//                    let rnd: u32 = rng.gen();
-//
-//                    let path_str = format!("database_{}.db", rnd);
-//                    let database_path = Path::new(path_str.as_str());
-//                    let rocks = Rocks::new(database_path.clone()).unwrap();
-//
-//                    // create new client and use different slabstore
-//                    let mut client =
-//                        GatewayClient::new("127.0.0.1:3333".parse().unwrap(), rocks).unwrap();
-//
-//                    // start client
-//                    client.start().await.unwrap();
-//
-//                    // sending slab
-//                    let _slab = Slab::new("testcoin".to_string(), rnd.to_le_bytes().to_vec());
-//                    client.put_slab(_slab).await.unwrap();
-//                })
-//            });
-//            thread_pools.push(thread);
-//        }
-//        for t in thread_pools {
-//            t.join().unwrap();
-//        }
-//    }
-//}
+//// $ cargo test test_ten_clients_simultaneously --bin darkfid
+//this will run 10 clients simultaneously
+
+//// $ cargo test test_subscriber --bin darkfid
+// Run Client A and send 10 slabs
+// Client B should receive 10 slabs from subscriber
+#[cfg(test)]
+mod test {
+
+    use std::net::SocketAddr;
+    use std::path::Path;
+    use std::sync::Arc;
+
+    use drk::blockchain::{rocks::columns, Rocks, RocksColumn, Slab};
+    use drk::service::{GatewayClient, GatewaySlabsSubscriber};
+
+    use async_executor::Executor;
+    use easy_parallel::Parallel;
+    use log::*;
+    use rand::Rng;
+    use simplelog::*;
+
+    pub async fn subscribe(gateway_slabs_sub: GatewaySlabsSubscriber, id: String) {
+        loop {
+            gateway_slabs_sub.recv().await.unwrap();
+            info!("Client {}: update state", id);
+        }
+    }
+
+    fn setup_log() {
+        let logger_config = ConfigBuilder::new().set_time_format_str("%T%.6f").build();
+
+        CombinedLogger::init(vec![
+            TermLogger::new(LevelFilter::Debug, logger_config, TerminalMode::Mixed).unwrap(),
+            WriteLogger::new(
+                LevelFilter::Debug,
+                Config::default(),
+                std::fs::File::create(Path::new("/tmp/dar.log")).unwrap(),
+            ),
+        ])
+        .unwrap();
+    }
+
+    #[test]
+    fn test_ten_clients_simultaneously() {
+        setup_log();
+
+        let mut thread_pools: Vec<std::thread::JoinHandle<()>> = vec![];
+
+        for _ in 0..10 {
+            let thread = std::thread::spawn(|| {
+                let ex = Arc::new(Executor::new());
+                let (signal, shutdown) = async_channel::unbounded::<()>();
+
+                let ex2 = ex.clone();
+
+                let (_, _) = Parallel::new()
+                    // Run four executor threads.
+                    .each(0..3, |_| smol::future::block_on(ex2.run(shutdown.recv())))
+                    // Run the main future on the current thread.
+                    .finish(|| {
+                        smol::future::block_on(async move {
+                            let connect_addr: SocketAddr = "127.0.0.1:3333".parse().unwrap();
+                            let sub_addr: SocketAddr = "127.0.0.1:4444".parse().unwrap();
+
+                            let mut rng = rand::thread_rng();
+                            let rnd: u32 = rng.gen();
+                            let path_str = format!("database_{}.db", rnd);
+
+                            let database_path = Path::new(path_str.as_str());
+                            let rocks = Rocks::new(database_path.clone()).unwrap();
+
+                            let slabstore = RocksColumn::<columns::Slabs>::new(rocks.clone());
+
+                            // create gateway client
+                            let mut client = GatewayClient::new(connect_addr, slabstore).unwrap();
+
+                            // start subscribing
+                            let gateway_slabs_sub: GatewaySlabsSubscriber =
+                                client.start_subscriber(sub_addr, ex.clone()).await.unwrap();
+                            ex.clone()
+                                .spawn(subscribe(gateway_slabs_sub, rnd.clone().to_string()))
+                                .detach();
+
+                            // start gateway client
+                            client.start().await.unwrap();
+
+                            let slab = Slab::new("btc".to_string(), rnd.to_le_bytes().to_vec());
+                            client.put_slab(slab).await.unwrap();
+                        });
+                        drop(signal);
+                        Ok::<(), drk::Error>(())
+                    });
+            });
+            thread_pools.push(thread);
+        }
+        for t in thread_pools {
+            t.join().unwrap();
+        }
+    }
+
+    #[test]
+    fn test_subscriber() {
+        setup_log();
+
+        let mut thread_pools: Vec<std::thread::JoinHandle<()>> = vec![];
+
+        // Client A
+        let thread = std::thread::spawn(|| {
+            smol::future::block_on(async move {
+                let connect_addr: SocketAddr = "127.0.0.1:3333".parse().unwrap();
+
+                let mut rng = rand::thread_rng();
+                let rnd: u32 = rng.gen();
+                let path_str = format!("database_{}.db", rnd);
+
+                let database_path = Path::new(path_str.as_str());
+                let rocks = Rocks::new(database_path.clone()).unwrap();
+
+                let slabstore = RocksColumn::<columns::Slabs>::new(rocks.clone());
+
+                // create gateway client
+                let mut client = GatewayClient::new(connect_addr, slabstore).unwrap();
+
+                // start gateway client
+                client.start().await.unwrap();
+
+                let slab = Slab::new("btc".to_string(), rnd.to_le_bytes().to_vec());
+
+                client.put_slab(slab.clone()).await.unwrap();
+                client.put_slab(slab.clone()).await.unwrap();
+                client.put_slab(slab.clone()).await.unwrap();
+                client.put_slab(slab.clone()).await.unwrap();
+                client.put_slab(slab.clone()).await.unwrap();
+                client.put_slab(slab.clone()).await.unwrap();
+                client.put_slab(slab.clone()).await.unwrap();
+                client.put_slab(slab.clone()).await.unwrap();
+                client.put_slab(slab.clone()).await.unwrap();
+                client.put_slab(slab.clone()).await.unwrap();
+            });
+        });
+        // Client B
+        let thread2 = std::thread::spawn(|| {
+            let ex = Arc::new(Executor::new());
+            let (signal, shutdown) = async_channel::unbounded::<()>();
+
+            let ex2 = ex.clone();
+
+            let (_, _) = Parallel::new()
+                // Run four executor threads.
+                .each(0..3, |_| smol::future::block_on(ex2.run(shutdown.recv())))
+                // Run the main future on the current thread.
+                .finish(|| {
+                    smol::future::block_on(async move {
+                        let connect_addr: SocketAddr = "127.0.0.1:3333".parse().unwrap();
+                        let sub_addr: SocketAddr = "127.0.0.1:4444".parse().unwrap();
+
+                        let mut rng = rand::thread_rng();
+                        let rnd: u32 = rng.gen();
+                        let path_str = format!("database_{}.db", rnd);
+
+                        let database_path = Path::new(path_str.as_str());
+                        let rocks = Rocks::new(database_path.clone()).unwrap();
+
+                        let slabstore = RocksColumn::<columns::Slabs>::new(rocks.clone());
+
+                        // create gateway client
+                        let mut client = GatewayClient::new(connect_addr, slabstore).unwrap();
+
+                        // start subscribing
+                        let gateway_slabs_sub: GatewaySlabsSubscriber =
+                            client.start_subscriber(sub_addr, ex.clone()).await.unwrap();
+
+                        ex.clone()
+                            .spawn(subscribe(gateway_slabs_sub, "B".to_string()))
+                            .detach();
+
+                        // start gateway client
+                        client.start().await.unwrap();
+
+                        // sleep for 2 seconds
+                        std::thread::sleep(std::time::Duration::from_secs(2));
+                    });
+                    drop(signal);
+                    Ok::<(), drk::Error>(())
+                });
+        });
+
+        thread_pools.push(thread);
+        thread_pools.push(thread2);
+
+        for t in thread_pools {
+            t.join().unwrap();
+        }
+    }
+}
--- a/src/bin/dfg.rs
+++ b/src/bin/dfg.rs
@@ -192,6 +192,7 @@ const VERTICES: &[model::ModelVertex] = &[
 //const INDICES: &[u16] = &[0, 1, 4, 1, 2, 4, 2, 3, 4];
 const INDICES: &[u16] = &[0, 1, 2, 0, 2, 3];

+#[allow(dead_code)]
 struct State {
    surface: wgpu::Surface,
    device: wgpu::Device,
--- a/src/bin/mint-classic.rs
+++ b/src/bin/mint-classic.rs
@@ -18,7 +18,7 @@ fn main() {

    {
        let params = setup_mint_prover();
-        save_params("mint.params", &params);
+        save_params("mint.params", &params).unwrap();
    }
    let (params, pvk) = load_params("mint.params").expect("params should load");

--- a/src/bin/mint.rs
+++ b/src/bin/mint.rs
@@ -25,7 +25,7 @@ fn unpack<F: PrimeField>(value: F) -> Vec<Scalar> {
 }

 // Unpack a u64 value in 64 Scalar binary digits
-fn unpack_u64(value: u64) -> Vec<Scalar> {
+fn _unpack_u64(value: u64) -> Vec<Scalar> {
    let mut result = Vec::with_capacity(64);

    for i in 0..64 {
@@ -56,7 +56,7 @@ fn main() -> Result<()> {
        visor.vm.constraints.len()
    );

-    visor.setup("mint.zts");
+    visor.setup("mint.zts")?;

    // We use the ExtendedPoint in calculations because it's faster
    let public_point = jubjub::ExtendedPoint::from(jubjub::SubgroupPoint::random(&mut OsRng));
--- a/src/bin/spend-classic.rs
+++ b/src/bin/spend-classic.rs
@@ -53,6 +53,7 @@ struct SpendRevealedValues {
    merkle_root: bls12_381::Scalar,
 }

+#[allow(dead_code)]
 impl SpendRevealedValues {
    fn compute(
        value: u64,
@@ -185,7 +186,7 @@ fn main() {

    {
        let params = setup_spend_prover();
-        save_params("spend.params", &params);
+        save_params("spend.params", &params).unwrap();
    }
    let (params, pvk) = load_params("spend.params").expect("params should load");

--- a/src/bin/tx-test.rs
+++ b/src/bin/tx-test.rs
@@ -1,29 +1,25 @@
-use async_std::sync;
 use bellman::groth16;
 use bls12_381::Bls12;
-use drk::{Error, Result};
-use ff::{Field, PrimeField};
-use log::*;
-use rand::rngs::OsRng;
+
 //use rocksdb::DB;
-use rusqlite::{named_params, Connection};
+
 //use std::fs::File;
-use rocksdb::{IteratorMode, Options, DB};
+
 use std::path::Path;

 use drk::crypto::{
    coin::Coin,
    load_params,
    merkle::{CommitmentTree, IncrementalWitness},
-    merkle_node::{hash_coin, MerkleNode},
+    merkle_node::MerkleNode,
    note::{EncryptedNote, Note},
    nullifier::Nullifier,
    save_params, setup_mint_prover, setup_spend_prover,
 };
-use drk::serial::{Decodable, Encodable};
-use drk::state::{state_transition, ProgramState, StateUpdate};
-use drk::wallet::walletdb::WalletDB;

+use drk::state::{ProgramState, StateUpdate};
+
+#[allow(dead_code)]
 struct MemoryState {
    // The entire merkle tree state
    tree: CommitmentTree<MerkleNode>,
@@ -52,7 +48,7 @@ struct MemoryState {

 impl ProgramState for MemoryState {
    // Vec<u8> for keys
-    fn is_valid_cashier_public_key(&self, public: &jubjub::SubgroupPoint) -> bool {
+    fn is_valid_cashier_public_key(&self, _public: &jubjub::SubgroupPoint) -> bool {
        //let path = WalletDB::wallet_path();
        //let connect = Connection::open(&path).expect("Failed to connect to database.");
        //let mut stmt = connect.prepare("SELECT key_public FROM keys").unwrap();
@@ -93,6 +89,7 @@ impl ProgramState for MemoryState {
    }
 }

+#[allow(dead_code)]
 impl MemoryState {
    fn apply(&mut self, mut update: StateUpdate) {
        // Extend our list of nullifiers with the ones from the update
@@ -135,7 +132,7 @@ impl MemoryState {
    }

    // sql
-    fn try_decrypt_note(&self, ciphertext: EncryptedNote) -> Option<(Note, jubjub::Fr)> {
+    fn try_decrypt_note(&self, _ciphertext: EncryptedNote) -> Option<(Note, jubjub::Fr)> {
        //debug!(target: "adapter", "try_decrypt_note() [START]");
        //let path = WalletDB::wallet_path();
        //debug!(target: "adapter", "try_decrypt_note() [FOUND PATH]");
@@ -176,8 +173,8 @@ fn main() {
    }

    // Load trusted setup parameters
-    let (mint_params, mint_pvk) = load_params("mint.params").expect("params should load");
-    let (spend_params, spend_pvk) = load_params("spend.params").expect("params should load");
+    let (_mint_params, _mint_pvk) = load_params("mint.params").expect("params should load");
+    let (_spend_params, _spend_pvk) = load_params("spend.params").expect("params should load");

    // Where is cashier private key stored? Does node have its own wallet schema
    // Cashier creates a secret key
--- a/src/bin/tx.rs
+++ b/src/bin/tx.rs
@@ -8,7 +8,7 @@ use drk::crypto::{
    coin::Coin,
    load_params,
    merkle::{CommitmentTree, IncrementalWitness},
-    merkle_node::{hash_coin, MerkleNode},
+    merkle_node::MerkleNode,
    note::{EncryptedNote, Note},
    nullifier::Nullifier,
    save_params, setup_mint_prover, setup_spend_prover,
@@ -119,11 +119,11 @@ fn main() {
    // Auto create trusted ceremony parameters if they don't exist
    if !Path::new("mint.params").exists() {
        let params = setup_mint_prover();
-        save_params("mint.params", &params);
+        save_params("mint.params", &params).unwrap();
    }
    if !Path::new("spend.params").exists() {
        let params = setup_spend_prover();
-        save_params("spend.params", &params);
+        save_params("spend.params", &params).unwrap();
    }

    // Load trusted setup parameters
@@ -183,11 +183,11 @@ fn main() {
    {
        // Here we simulate 5 fake random coins, adding them to our tree.
        let tree = &mut state.tree;
-        for i in 0..5 {
+        for _i in 0..5 {
            // Don't worry about any of the code in this block
            // We're just filling the tree with fake coins
            let cmu = MerkleNode::new(bls12_381::Scalar::random(&mut OsRng).to_repr());
-            tree.append(cmu);
+            tree.append(cmu).unwrap();

            let root = tree.root();
            state.merkle_roots.push(root.into());
@@ -218,12 +218,12 @@ fn main() {
        assert_eq!(tree.root(), witness.root());

        // Add some more random coins in
-        for i in 0..10 {
+        for _i in 0..10 {
            // Don't worry about any of the code in this block
            // We're just filling the tree with fake coins
            let cmu = MerkleNode::new(bls12_381::Scalar::random(&mut OsRng).to_repr());
-            tree.append(cmu);
-            witness.append(cmu);
+            tree.append(cmu).unwrap();
+            witness.append(cmu).unwrap();
            assert_eq!(tree.root(), witness.root());

            let root = tree.root();
--- a/src/blockchain/slab.rs
+++ b/src/blockchain/slab.rs
@@ -1,6 +1,7 @@
 use crate::serial::{Decodable, Encodable};
 use crate::Result;

+#[derive(Clone, Debug)]
 pub struct Slab {
    asset_type: String,
    index: u64,
--- a/src/crypto/coin.rs
+++ b/src/crypto/coin.rs
@@ -16,13 +16,13 @@ impl Coin {
 }

 impl Encodable for Coin {
-    fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
+    fn encode<S: io::Write>(&self, s: S) -> Result<usize> {
        Ok(self.repr.encode(s)?)
    }
 }

 impl Decodable for Coin {
-    fn decode<D: io::Read>(mut d: D) -> Result<Self> {
+    fn decode<D: io::Read>(d: D) -> Result<Self> {
        Ok(Self {
            repr: Decodable::decode(d)?,
        })
--- a/src/crypto/merkle_node.rs
+++ b/src/crypto/merkle_node.rs
@@ -119,13 +119,13 @@ impl From<MerkleNode> for bls12_381::Scalar {
 }

 impl Encodable for MerkleNode {
-    fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
+    fn encode<S: io::Write>(&self, s: S) -> Result<usize> {
        Ok(self.repr.encode(s)?)
    }
 }

 impl Decodable for MerkleNode {
-    fn decode<D: io::Read>(mut d: D) -> Result<Self> {
+    fn decode<D: io::Read>(d: D) -> Result<Self> {
        Ok(Self {
            repr: Decodable::decode(d)?,
        })
--- a/src/crypto/nullifier.rs
+++ b/src/crypto/nullifier.rs
@@ -16,13 +16,13 @@ impl Nullifier {
 }

 impl Encodable for Nullifier {
-    fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
+    fn encode<S: io::Write>(&self, s: S) -> Result<usize> {
        Ok(self.repr.encode(s)?)
    }
 }

 impl Decodable for Nullifier {
-    fn decode<D: io::Read>(mut d: D) -> Result<Self> {
+    fn decode<D: io::Read>(d: D) -> Result<Self> {
        Ok(Self {
            repr: Decodable::decode(d)?,
        })
--- a/src/old/mimc.rs
+++ b/src/old/mimc.rs
@@ -56,19 +56,19 @@ fn mimc<Scalar: PrimeField>(mut xl: Scalar, mut xr: Scalar, constants: &[Scalar]
    xl
 }

-macro_rules! from_slice {
-    ($data:expr, $len:literal) => {{
-        let mut array = [0; $len];
-        // panics if not enough data
-        let bytes = &$data[..array.len()];
-        assert_eq!(bytes.len(), array.len());
-        for (a, b) in array.iter_mut().rev().zip(bytes.iter()) {
-            *a = *b;
-        }
-        //array.copy_from_slice(bytes.iter().rev());
-        array
-    }};
-}
+//macro_rules! from_slice {
+//    ($data:expr, $len:literal) => {{
+//        let mut array = [0; $len];
+//        // panics if not enough data
+//        let bytes = &$data[..array.len()];
+//        assert_eq!(bytes.len(), array.len());
+//        for (a, b) in array.iter_mut().rev().zip(bytes.iter()) {
+//            *a = *b;
+//        }
+//        //array.copy_from_slice(bytes.iter().rev());
+//        array
+//    }};
+//}

 /// This is our demo circuit for proving knowledge of the
 /// preimage of a MiMC hash invocation.
--- a/src/rpc/adapter.rs
+++ b/src/rpc/adapter.rs
@@ -1,7 +1,7 @@
 use crate::wallet::WalletDB;
-use crate::{Error, Result};
+use crate::Result;
 use log::*;
-use std::path::{Path, PathBuf};
+use std::path::PathBuf;
 use std::sync::Arc;

 // Dummy adapter for now
--- a/src/rpc/jsonserver.rs
+++ b/src/rpc/jsonserver.rs
@@ -101,6 +101,7 @@ pub async fn start(
    Ok(())
 }
 // json RPC server goes here
+#[allow(dead_code)]
 pub struct RpcInterface {
    p2p: Arc<net::P2p>,
    pub started: Mutex<bool>,
--- a/src/service/gateway.rs
+++ b/src/service/gateway.rs
@@ -8,7 +8,7 @@ use crate::{serial::deserialize, serial::serialize, Error, Result};
 use async_executor::Executor;
 use log::*;

-pub type Slabs = Vec<Vec<u8>>;
+pub type GatewaySlabsSubscriber = async_channel::Receiver<Slab>;

 #[repr(u8)]
 enum GatewayError {
@@ -179,6 +179,8 @@ impl GatewayService {
 pub struct GatewayClient {
    protocol: ReqProtocol,
    slabstore: Arc<SlabStore>,
+    gateway_slabs_sub_s: async_channel::Sender<Slab>,
+    gateway_slabs_sub_rv: GatewaySlabsSubscriber,
 }

 impl GatewayClient {
@@ -187,9 +189,13 @@ impl GatewayClient {

        let slabstore = SlabStore::new(rocks)?;

+        let (gateway_slabs_sub_s, gateway_slabs_sub_rv) = async_channel::unbounded::<Slab>();
+
        Ok(GatewayClient {
            protocol,
            slabstore,
+            gateway_slabs_sub_s,
+            gateway_slabs_sub_rv,
        })
    }

@@ -211,7 +217,6 @@ impl GatewayClient {
        if last_index > 0 {
            for index in (local_last_index + 1)..(last_index + 1) {
                if let None = self.get_slab(index).await? {
-                    warn!("Index not exist");
                    break;
                }
            }
@@ -221,14 +226,20 @@ impl GatewayClient {
        Ok(last_index)
    }

-    pub async fn get_slab(&mut self, index: u64) -> Result<Option<Vec<u8>>> {
+    pub async fn get_slab(&mut self, index: u64) -> Result<Option<Slab>> {
        let rep = self
            .protocol
-            .request(GatewayCommand::GetSlab as u8, serialize(&index))
+            .request(
+                GatewayCommand::GetSlab as u8,
+                serialize(&index),
+                &handle_error,
+            )
            .await?;

        if let Some(slab) = rep {
-            self.slabstore.put(deserialize(&slab)?)?;
+            let slab: Slab = deserialize(&slab)?;
+            self.gateway_slabs_sub_s.send(slab.clone()).await?;
+            self.slabstore.put(slab.clone())?;
            return Ok(Some(slab));
        }
        Ok(None)
@@ -242,7 +253,7 @@ impl GatewayClient {

            let rep = self
                .protocol
-                .request(GatewayCommand::PutSlab as u8, slab.clone())
+                .request(GatewayCommand::PutSlab as u8, slab.clone(), &handle_error)
                .await?;

            if let Some(_) = rep {
@@ -255,7 +266,7 @@ impl GatewayClient {
    pub async fn get_last_index(&mut self) -> Result<u64> {
        let rep = self
            .protocol
-            .request(GatewayCommand::GetLastIndex as u8, vec![])
+            .request(GatewayCommand::GetLastIndex as u8, vec![], &handle_error)
            .await?;
        if let Some(index) = rep {
            return Ok(deserialize(&index)?);
@@ -267,9 +278,44 @@ impl GatewayClient {
        self.slabstore.clone()
    }

-    pub async fn start_subscriber(sub_addr: SocketAddr) -> Result<Subscriber> {
+    pub async fn start_subscriber(
+        &self,
+        sub_addr: SocketAddr,
+        executor: Arc<Executor<'_>>,
+    ) -> Result<GatewaySlabsSubscriber> {
        let mut subscriber = Subscriber::new(sub_addr, String::from("GATEWAY CLIENT"));
        subscriber.start().await?;
-        Ok(subscriber)
+        executor
+            .spawn(Self::subscribe_loop(
+                subscriber,
+                self.slabstore.clone(),
+                self.gateway_slabs_sub_s.clone(),
+            ))
+            .detach();
+        Ok(self.gateway_slabs_sub_rv.clone())
+    }
+
+    async fn subscribe_loop(
+        mut subscriber: Subscriber,
+        slabstore: Arc<SlabStore>,
+        gateway_slabs_sub_s: async_channel::Sender<Slab>,
+    ) -> Result<()> {
+        loop {
+            let slab = subscriber.fetch::<Slab>().await?;
+            gateway_slabs_sub_s.send(slab.clone()).await?;
+            slabstore.put(slab)?;
+        }
+    }
+}
+
+fn handle_error(status_code: u32) {
+    match status_code {
+        1 => {
+            warn!("Reply has an Error: Index is not updated");
+        }
+        2 => {
+            warn!("Reply has an Error: Index Not Exist");
+        }
+        _ => {}
    }
 }
--- a/src/service/mod.rs
+++ b/src/service/mod.rs
@@ -2,6 +2,5 @@ pub mod gateway;
 pub mod options;
 pub mod reqrep;

-pub use gateway::{GatewayClient, GatewayService};
+pub use gateway::{GatewayClient, GatewayService, GatewaySlabsSubscriber};
 pub use options::{ClientProgramOptions, ProgramOptions};
-pub use reqrep::Subscriber;
--- a/src/service/reqrep.rs
+++ b/src/service/reqrep.rs
@@ -144,7 +144,12 @@ impl ReqProtocol {
        Ok(())
    }

-    pub async fn request(&mut self, command: u8, data: Vec<u8>) -> Result<Option<Vec<u8>>> {
+    pub async fn request(
+        &mut self,
+        command: u8,
+        data: Vec<u8>,
+        handle_error: &dyn Fn(u32),
+    ) -> Result<Option<Vec<u8>>> {
        let request = Request::new(command, data);
        let req = serialize(&request);
        let req = bytes::Bytes::from(req);
@@ -168,9 +173,10 @@ impl ReqProtocol {
                reply.has_error()
            );

-            // TODO return error status code instead of None
            if reply.has_error() {
-                warn!("Reply has an error {}", reply.get_error());
+                // TODO return error status code instead of None
+                // this is temporary
+                handle_error(reply.get_error());
                return Ok(None);
            }

--- a/src/wallet/walletdb.rs
+++ b/src/wallet/walletdb.rs
@@ -60,7 +60,7 @@ impl WalletDB {
    pub async fn get(path: PathBuf) -> Result<()> {
        debug!(target: "get_cash_public", "Returning cashier keys...");
        let connect = Connection::open(&path).expect("Failed to connect to database.");
-        let id = 0;
+        let _id = 0;
        let mut stmt = connect.prepare("SELECT key_public FROM keys").unwrap();
        let key_iter = stmt
            .query_map::<Vec<u8>, _, _>([], |row| row.get(0))
@@ -92,5 +92,3 @@ impl WalletDB {
        Ok(())
    }
 }
-
-fn main() {}