v7/depth view (#124)

* add orderbook depth view * publish orderbook to sidecar * close orderbook :( * update CHANGELOG * enable migration feature * including rustls feature * online test of migration --------- Co-authored-by: Cyberaurora <[email protected]>
uinb · Dec 14, 2023 · 87aee02 · 87aee02
1 parent e3aaeda
commit 87aee02
Show file tree

Hide file tree

Showing 14 changed files with 335 additions and 149 deletions.
diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
@@ -29,7 +29,7 @@ jobs:
         uses: actions-rs/cargo@v1
         with:
           command: build
-          args: --release
+          args: --release --all-features
       - uses: montudor/action-zip@v1
         with:
           args: zip -j -r galois.zip target/release/galois target/release/sidecar galois.toml.example sidecar.toml.example README.md LICENSE

diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,3 +1,7 @@
+# v0.7.0-rc.13
+
+- online test of migration
+
 # v0.5.1
 
 - enable polygon

diff --git a/Cargo.lock b/Cargo.lock
diff --git a/engine/Cargo.toml b/engine/Cargo.toml
@@ -9,7 +9,7 @@ description = "High performance matching system"
 
 [features]
 default = []
-v1-to-v2 = ["sqlx"]
+v1-to-v2 = ["sqlx", "tokio"]
 
 [dependencies]
 rust_decimal = { version = "1.22", features = ["serde-bincode"] }
@@ -20,7 +20,8 @@ async-trait = "0.1.63"
 serde_json = "1.0"
 rocksdb = "0.21"
 flate2 = { version = "1.0", features = ["zlib"], default-features = false }
-sqlx = { version = "0.6.2", features = ["mysql", "decimal", "chrono"], optional = true }
+sqlx = { version = "0.6.2", features = ["mysql", "decimal", "chrono", "runtime-tokio-rustls", "time"], optional = true }
+tokio = { version = "1", features = ["full"], optional = true }
 toml = "0.5"
 lazy_static = "1.4"
 linked-hash-map = { version = "0.5.3", features = ["serde_impl"] }
@@ -45,6 +46,7 @@ memmap = "0.7"
 dashmap = "5.4.0"
 indexmap = "1.9.2"
 rand = "0.8.5"
+signal-hook = "0.3"
 smt = { git = "https://github.com/uinb/sparse-merkle-tree", tag = "v0.1.8", package = "sparse-merkle-tree", features = ["serde-rs", "blake2b"] }
 sub-api = { package = "substrate-api-client", git = "https://github.com/uinb/fusotao-rust-client.git", branch = "master" }
 node-api = { package = "ac-node-api", git = "https://github.com/uinb/fusotao-rust-client.git", branch = "master" }

diff --git a/engine/src/config.rs b/engine/src/config.rs
@@ -68,7 +68,7 @@ pub struct MigrateCmd {
         help = "The old coredump file path"
     )]
     pub input_path: String,
-    #[arg(long, action=clap::ArgAction::SetFalse, help = "Migrate coredump file only if set")]
+    #[arg(long, action=clap::ArgAction::SetTrue, help = "Migrate coredump file only if set")]
     pub core_only: bool,
 }
 

diff --git a/engine/src/executor/mod.rs b/engine/src/executor/mod.rs
@@ -20,9 +20,9 @@ pub mod orders;
 
 use crate::{
     core::*,
-    input::{Event, Message},
+    input::{self, Event, Message},
     orderbook::*,
-    output::Output,
+    output::{Depth, Output},
     prover, snapshot,
 };
 use anyhow::anyhow;
@@ -64,7 +64,7 @@ pub fn init(recv: DriverChannel, market: MarketChannel, response: ResponseChanne
                     log::debug!("event {} rejected: {}", id, e);
                     let msg = json!({"error": e.to_string()});
                     let v = to_vec(&msg).unwrap_or_default();
-                    let _ = response.send((session, Message::new(req_id, v)));
+                    let _ = response.send((session, Message::new_req(req_id, v)));
                 }
                 Err(EventsError::EventIgnored(id, e)) => {
                     log::info!("event {} ignored: {}", id, e);
@@ -135,11 +135,12 @@ fn do_execute(
                 quote_fee: Decimal::zero(),
             });
             // compatiable with old version since we don't use mysql auto increment id anymore
+            // session=0 indicates replaying from snapshot
             if session != 0 {
                 response
                     .send((
                         session,
-                        Message::new(
+                        Message::new_req(
                             req_id,
                             to_vec(&json!({
                                 "id": mr.taker.order_id
@@ -148,6 +149,17 @@ fn do_execute(
                         ),
                     ))
                     .map_err(|_| EventsError::Interrupted(id))?;
+                let orderbook: &_ = orderbook;
+                let depth: Depth = (cmd.symbol, orderbook).into();
+                response
+                    .send((
+                        0,
+                        Message::new_broadcast(
+                            input::DEPTH_UPDATED,
+                            to_vec(&depth).unwrap_or_default(),
+                        ),
+                    ))
+                    .map_err(|_| EventsError::Interrupted(id))?;
             }
             let out = clearing::clear(
                 &mut data.accounts,
@@ -163,7 +175,13 @@ fn do_execute(
                 if session != 0 {
                     // broadcast to all sessions
                     response
-                        .send((0, Message::new(0, to_vec(&o).unwrap_or_default())))
+                        .send((
+                            0,
+                            Message::new_broadcast(
+                                input::ORDER_MATCHED,
+                                to_vec(&o).unwrap_or_default(),
+                            ),
+                        ))
                         .map_err(|_| EventsError::Interrupted(id))?;
                 }
             }
@@ -231,7 +249,7 @@ fn do_execute(
                 response
                     .send((
                         session,
-                        Message::new(
+                        Message::new_req(
                             req_id,
                             to_vec(&json!({
                                 "id": cmd.order_id
@@ -240,6 +258,17 @@ fn do_execute(
                         ),
                     ))
                     .map_err(|_| EventsError::Interrupted(id))?;
+                let orderbook: &_ = orderbook;
+                let depth: Depth = (cmd.symbol, orderbook).into();
+                response
+                    .send((
+                        0,
+                        Message::new_broadcast(
+                            input::DEPTH_UPDATED,
+                            to_vec(&depth).unwrap_or_default(),
+                        ),
+                    ))
+                    .map_err(|_| EventsError::Interrupted(id))?;
             }
             let out = clearing::clear(
                 &mut data.accounts,
@@ -398,42 +427,46 @@ fn do_execute(
                     .map_or(vec![], |order| to_vec(order).unwrap_or_default()),
                 None => vec![],
             };
-            let _ = response.send((session, Message::new(req_id, v)));
+            let _ = response.send((session, Message::new_req(req_id, v)));
             Ok(())
         }
         Event::QueryUserOrders(symbol, user_id, session, req_id) => {
             let o = data.orders.list(user_id, symbol);
             let v = to_vec(&o).unwrap_or_default();
-            let _ = response.send((session, Message::new(req_id, v)));
+            let _ = response.send((session, Message::new_req(req_id, v)));
             Ok(())
         }
         Event::QueryBalance(user_id, currency, session, req_id) => {
             let a = assets::get_balance_to_owned(&data.accounts, &user_id, currency);
             let v = to_vec(&a).unwrap_or_default();
-            let _ = response.send((session, Message::new(req_id, v)));
+            let _ = response.send((session, Message::new_req(req_id, v)));
             Ok(())
         }
         Event::QueryAccounts(user_id, session, req_id) => {
             let a = assets::get_account_to_owned(&data.accounts, &user_id);
             let v = to_vec(&a).unwrap_or_default();
-            let _ = response.send((session, Message::new(req_id, v)));
+            let _ = response.send((session, Message::new_req(req_id, v)));
             Ok(())
         }
         Event::QueryExchangeFee(symbol, session, req_id) => {
-            let mut v: HashMap<String, Fee> = HashMap::new();
-            let orderbook = data.orderbooks.get(&symbol);
-            match orderbook {
-                Some(book) => {
-                    v.insert(String::from("maker_fee"), book.maker_fee);
-                    v.insert(String::from("taker_fee"), book.taker_fee);
-                }
-                _ => {
-                    v.insert(String::from("maker_fee"), Decimal::new(0, 0));
-                    v.insert(String::from("taker_fee"), Decimal::new(0, 0));
-                }
-            }
-            let v = to_vec(&v).unwrap_or_default();
-            let _ = response.send((session, Message::new(req_id, v)));
+            let (maker, taker) = data
+                .orderbooks
+                .get(&symbol)
+                .map(|b| (b.maker_fee, b.taker_fee))
+                .unwrap_or_default();
+            let map = HashMap::from([("maker_fee", maker), ("taker_fee", taker)]);
+            let v = to_vec(&map).unwrap_or_default();
+            let _ = response.send((session, Message::new_req(req_id, v)));
+            Ok(())
+        }
+        Event::QueryAllOrderbooks(session, req_id) => {
+            let depth = data
+                .orderbooks
+                .iter()
+                .map(|(s, o)| (*s, o).into())
+                .collect::<Vec<Depth>>();
+            let v = to_vec(&depth).unwrap_or_default();
+            let _ = response.send((session, Message::new_req(req_id, v)));
             Ok(())
         }
         Event::Dump(id) => {