Implement trigger for special orders

Author: cherryman

Cargo.lock

@@ -2,7 +2,7 @@
 name = "zo-keeper"
 version = "0.1.0"
 edition = "2021"
-rust-version = "1.58"
+rust-version = "1.63"
 
 [features]
 devnet = ["zo-abi/devnet"]
@@ -37,3 +37,4 @@ num-traits = "0.2"
 fixed = "1"
 serum_dex = "0.5"
 spl-token = "3.2"
+parking_lot = "0.12"

Cargo.toml

@@ -1,7 +1,8 @@
 FROM rust:latest
 RUN apt-get update \
     && apt-get install -y libudev-dev libclang-dev lld cmake \
-    && rustup component add rustfmt
+    && rustup component add rustfmt \
+    && rustup default nightly
 WORKDIR /srv
 COPY . .
 RUN cargo build --release \

Dockerfile

@@ -1,3 +1,4 @@
 crank: ./target/release/zo-keeper crank
 consumer: ./target/release/zo-keeper consumer
 recorder: ./target/release/zo-keeper recorder
+trigger: ./target/release/zo-keeper trigger

Procfile

@@ -9,4 +9,5 @@ case "$DYNO" in
     consumer.*) run consumer ;;
     liquidator.*) run liquidator ;;
     recorder.*) run recorder ;;
+    trigger.*) run trigger ;;
 esac

abi

@@ -0,0 +1,41 @@
+use anchor_client::{
+    solana_client::{
+        nonblocking::pubsub_client::PubsubClient,
+        rpc_config::{RpcTransactionLogsConfig, RpcTransactionLogsFilter},
+        rpc_response::RpcLogsResponse,
+    },
+    solana_sdk::commitment_config::CommitmentConfig,
+};
+use futures::StreamExt as _;
+use zo_abi as zo;
+
+pub async fn run(st: &'static crate::AppState) -> Result<(), crate::Error> {
+    let cli = PubsubClient::new(st.cluster.ws_url()).await.unwrap();
+    let (mut rx, _unsub) = cli
+        .logs_subscribe(
+            RpcTransactionLogsFilter::Mentions(vec![zo::ID.to_string()]),
+            RpcTransactionLogsConfig {
+                commitment: Some(CommitmentConfig::finalized()),
+            },
+        )
+        .await
+        .unwrap();
+
+    while let Some(x) = rx.next().await {
+        tokio::task::spawn_blocking(move || print_log(&x.value));
+    }
+
+    Ok(())
+}
+
+fn print_log(r: &RpcLogsResponse) {
+    let status = if r.err.is_none() { "ok " } else { "err" };
+
+    let mut i = 0;
+    for l in r.logs.iter() {
+        if let Some(ix) = l.strip_prefix("Program log: Instruction: ") {
+            tracing::info!("ix: {} {} [{}] {}", status, r.signature, i, ix);
+            i += 1;
+        }
+    }
+}

scripts/heroku-docker-cmd.sh

@@ -11,6 +11,8 @@ pub enum Error {
     #[error("{0}: {0:?}")]
     SolanaClient(#[from] solana_client::client_error::ClientError),
     #[error("{0}: {0:?}")]
+    SolanaPubsubClient(#[from] solana_client::pubsub_client::PubsubClientError),
+    #[error("{0}: {0:?}")]
     JsonRpc(#[from] jsonrpc_core_client::RpcError),
     #[error("{0}: {0:?}")]
     TransactionError(

src/custom.rs

@@ -2,6 +2,7 @@ pub mod consumer;
 pub mod crank;
 pub mod liquidator;
 pub mod recorder;
+pub mod trigger;
 
 mod db;
 mod error;

src/error.rs

@@ -74,6 +74,9 @@ enum Command {
 
     /// Listen and store events into a database
     Recorder,
+
+    /// Trigger special orders.
+    Trigger,
 }
 
 fn main() -> Result<(), lib::Error> {
@@ -90,6 +93,16 @@ fn main() -> Result<(), lib::Error> {
             .init();
     }
 
+    {
+        // Ensure that a panic in a spawned thread exits the main process.
+        // Unfortunately, other threads' resources are not necessarily freed.
+        let hook = std::panic::take_hook();
+        std::panic::set_hook(Box::new(move |x| {
+            hook(x);
+            std::process::exit(255);
+        }));
+    }
+
     let Cli {
         rpc_url,
         ws_url,
@@ -160,6 +173,7 @@ fn main() -> Result<(), lib::Error> {
             },
         ))?,
         Command::Recorder => rt.block_on(lib::recorder::run(app_state))?,
+        Command::Trigger => lib::trigger::run(app_state)?,
     };
 
     Ok(())