final 0.9.1 rc tweaks

kinode/Cargo.toml

@@ -26,7 +26,7 @@ simulation-mode = []
 
 [dependencies]
 aes-gcm = "0.10.3"
-alloy = { git = "https://github.com/bitful-pannul/alloy.git", rev = "bd900b4", features = [
+alloy = { git = "https://github.com/kinode-dao/alloy.git", rev = "e672f3e", features = [
     "consensus",
     "contract",
     "json-rpc",

kinode/packages/app_store/chain/src/lib.rs

@@ -45,7 +45,7 @@ const KIMAP_FIRST_BLOCK: u64 = kimap::KIMAP_FIRST_BLOCK;
 #[cfg(feature = "simulation-mode")]
 const KIMAP_FIRST_BLOCK: u64 = 1;
 
-const DELAY_MS: u64 = 5_000;
+const DELAY_MS: u64 = 1_000; // 1s
 
 #[derive(Debug, Serialize, Deserialize)]
 pub struct State {

kinode/packages/kns_indexer/kns_indexer/src/lib.rs

@@ -37,7 +37,7 @@ const KIMAP_FIRST_BLOCK: u64 = 1; // local
 
 const MAX_PENDING_ATTEMPTS: u8 = 3;
 const SUBSCRIPTION_TIMEOUT: u64 = 60;
-const NEW_BLOCK_TICK: u64 = 3000; // 3s
+const DELAY_MS: u64 = 1_000; // 1s
 
 #[derive(Clone, Debug, Serialize, Deserialize)]
 struct State {
@@ -236,10 +236,7 @@ fn handle_eth_message(
             }
         }
         Ok(Err(e)) => {
-            print_to_terminal(
-                0,
-                &format!("got eth subscription error ({e:?}), resubscribing"),
-            );
+            println!("got eth subscription error ({e:?}), resubscribing");
             if e.id == 1 {
                 eth_provider.subscribe_loop(1, mints_filter.clone());
             } else if e.id == 2 {
@@ -251,14 +248,14 @@ fn handle_eth_message(
     if tick {
         let block_number = eth_provider.get_block_number();
         if let Ok(block_number) = block_number {
-            print_to_terminal(1, &format!("new block: {}", block_number));
+            print_to_terminal(2, &format!("new block: {}", block_number));
             state.last_block = block_number;
         }
     }
     handle_pending_notes(state, pending_notes)?;
 
     if !pending_notes.is_empty() {
-        timer::set_timer(NEW_BLOCK_TICK, None);
+        timer::set_timer(DELAY_MS, None);
     }
 
     Ok(())
@@ -279,10 +276,7 @@ fn handle_pending_notes(
             for (note, attempt) in notes.drain(..) {
                 if attempt >= MAX_PENDING_ATTEMPTS {
                     // skip notes that have exceeded max attempts
-                    print_to_terminal(
-                        1,
-                        &format!("dropping note from block {block} after {attempt} attempts"),
-                    );
+                    println!("dropping note from block {block} after {attempt} attempts");
                     continue;
                 }
                 if let Err(e) = handle_note(state, &note) {
@@ -441,7 +435,7 @@ fn handle_log(
             }
             if let Some(block_number) = log.block_number {
                 print_to_terminal(
-                    0,
+                    1,
                     &format!("adding note to pending_notes for block {block_number}"),
                 );
                 pending_notes

kinode/src/eth/subscription.rs

@@ -92,6 +92,7 @@ pub async fn create_new_subscription(
                                         chain_id,
                                         &providers,
                                         close_receiver,
+                                        &print_tx,
                                     )
                                     .await;
                                     let Err(e) = r else {
@@ -257,8 +258,6 @@ async fn build_subscription(
                     )
                     .await;
                 }
-                let alloy_sub_id = rx.local_id();
-                let alloy_sub_id: alloy::primitives::U256 = alloy_sub_id.clone().into();
                 return Ok(Ok((rx, chain_id)));
             }
             Err(rpc_error) => {
@@ -385,11 +384,12 @@ async fn maintain_local_subscription(
     chain_id: u64,
     providers: &Providers,
     mut close_receiver: tokio::sync::mpsc::Receiver<bool>,
+    print_tx: &PrintSender,
 ) -> Result<(), EthSubError> {
     loop {
         tokio::select! {
             _ = close_receiver.recv() => {
-                unsubscribe(rx, &chain_id, providers);
+                unsubscribe(rx, &chain_id, providers, print_tx).await;
                 return Ok(());
             },
             value = rx.recv() => {
@@ -424,14 +424,19 @@ async fn maintain_local_subscription(
         .and_modify(|sub_map| {
             sub_map.remove(&sub_id);
         });
-    unsubscribe(rx, &chain_id, providers);
+    unsubscribe(rx, &chain_id, providers, print_tx).await;
     Err(EthSubError {
         id: sub_id,
         error: format!("subscription ({target}) closed unexpectedly"),
     })
 }
 
-fn unsubscribe(rx: RawSubscription, chain_id: &u64, providers: &Providers) {
+async fn unsubscribe(
+    rx: RawSubscription,
+    chain_id: &u64,
+    providers: &Providers,
+    print_tx: &PrintSender,
+) {
     let alloy_sub_id = rx.local_id();
     let alloy_sub_id = alloy_sub_id.clone().into();
     let Some(chain_providers) = providers.get_mut(chain_id) else {
@@ -441,7 +446,14 @@ fn unsubscribe(rx: RawSubscription, chain_id: &u64, providers: &Providers) {
         let Some(pubsub) = url.pubsub.as_ref() else {
             continue;
         };
-        let x = pubsub.unsubscribe(alloy_sub_id);
+        if let Err(err) = pubsub.unsubscribe(alloy_sub_id) {
+            let _ = print_tx
+                .send(Printout {
+                    verbosity: 0,
+                    content: format!("unsubscribe from ETH RPC failed: {err:?}"),
+                })
+                .await;
+        }
     }
 }
 

lib/Cargo.toml

@@ -15,7 +15,7 @@ kit = { git = "https://github.com/kinode-dao/kit", tag = "v0.6.8" }
 tokio = "1.28"
 
 [dependencies]
-alloy = { git = "https://github.com/bitful-pannul/alloy.git", rev = "bd900b4", features = [
+alloy = { git = "https://github.com/kinode-dao/alloy.git", rev = "e672f3e", features = [
     "json-rpc",
     "rpc-types",
     "rpc-types-eth",