Skip to content

Commit bb44ae3

Browse files
imotaiimotai
authored
fix: handle the arweave-rs exception (#598)
* fix: handle the arweave-rs exception * fix: revert the client_v2.test.ts * fix: fix the empty log bugs
1 parent 518b627 commit bb44ae3

File tree

7 files changed

+217
-170
lines changed

7 files changed

+217
-170
lines changed

‎Cargo.toml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,6 @@ fastcrypto = {git = "https://github.com/MystenLabs/fastcrypto", rev = "306465d4f
1616
ethers = {git="https://github.com/imotai/ethers-rs", rev="d526191b7972e8cf4412fee8b71cbf42e0ce7995"}
1717
tonic = {git="https://github.com/hyperium/tonic", rev="ae7580160431cd25c1eecda4c85014ef6ce8d93f"}
1818
tonic-web = {git="https://github.com/hyperium/tonic", rev="ae7580160431cd25c1eecda4c85014ef6ce8d93f"}
19-
arweave-rs = {git="https://github.com/imotai/arweave-rs", rev="c46183c4cef05004685e53a869606d18abcbcdf2"}
19+
arweave-rs = {git="https://github.com/imotai/arweave-rs", rev="ce7bcaadc614afb0e991953da6861ff173dce6ea"}
2020
ejdb2rs = {git="https://github.com/imotai/ejdb2rs", rev="dfe9a00c66986c072b8375ce68086b9c98ded4fe"}
2121
serde_json= "1.0"

‎sdk/src/index.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -59,7 +59,7 @@ export {
5959
getDatabase,
6060
getCollection,
6161
addIndex,
62-
deleteEventDatabase
62+
deleteEventDatabase,
6363
} from './store/database_v2'
6464

6565
export { Index, IndexType } from './proto/db3_database_v2'

‎src/event/src/event_processor.rs‎

Lines changed: 87 additions & 71 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ use ethers::{
2929
use std::collections::HashSet;
3030
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
3131
use std::sync::Arc;
32+
use tokio::time::{sleep, Duration};
3233
use tracing::{info, warn};
3334

3435
#[derive(Debug)]
@@ -43,7 +44,6 @@ pub struct EventProcessorConfig {
4344

4445
pub struct EventProcessor {
4546
config: EventProcessorConfig,
46-
provider: Arc<Provider<Ws>>,
4747
running: Arc<AtomicBool>,
4848
db_store: DBStoreV2,
4949
block_number: Arc<AtomicU64>,
@@ -56,13 +56,8 @@ unsafe impl Send for EventProcessor {}
5656
impl EventProcessor {
5757
pub async fn new(config: EventProcessorConfig, db_store: DBStoreV2) -> Result<Self> {
5858
info!("new event processor with config {:?}", config);
59-
let provider = Provider::<Ws>::connect(&config.evm_node_url)
60-
.await
61-
.map_err(|e| DB3Error::StoreEventError(format!("{e}")))?;
62-
let provider_arc = Arc::new(provider);
6359
Ok(Self {
6460
config,
65-
provider: provider_arc,
6661
running: Arc::new(AtomicBool::new(false)),
6762
db_store,
6863
block_number: Arc::new(AtomicU64::new(0)),
@@ -91,8 +86,6 @@ impl EventProcessor {
9186
}
9287

9388
pub async fn start(&self) -> Result<()> {
94-
let abi: Abi = serde_json::from_str(self.config.abi.as_str())
95-
.map_err(|e| DB3Error::StoreEventError(format!("{e}")))?;
9689
self.running
9790
.store(true, std::sync::atomic::Ordering::Relaxed);
9891
let address = self
@@ -102,88 +95,111 @@ impl EventProcessor {
10295
.map_err(|e| DB3Error::StoreEventError(format!("{e}")))?;
10396
let db_addr = DB3Address::from_hex(self.config.db_addr.as_str())
10497
.map_err(|e| DB3Error::StoreEventError(format!("{e}")))?;
105-
let filter = match self.config.start_block == 0 {
106-
true => Filter::new().address(address),
107-
false => {
108-
info!(
109-
"start process contract from block {} with address {}",
110-
self.config.start_block, self.config.contract_addr
111-
);
112-
Filter::new()
113-
.from_block(self.config.start_block)
114-
.address(address)
115-
}
116-
};
117-
let mut stream = self
118-
.provider
119-
.subscribe_logs(&filter)
120-
.await
98+
let abi: Abi = serde_json::from_str(self.config.abi.as_str())
12199
.map_err(|e| DB3Error::StoreEventError(format!("{e}")))?;
122-
info!(
123-
"event processor for contract {}",
124-
self.config.contract_addr.as_str()
125-
);
126-
while let Some(log) = stream.next().await {
100+
let local_evm_node_url = self.config.evm_node_url.to_string();
101+
loop {
127102
if !self.running.load(Ordering::Relaxed) {
128103
info!(
129104
"stop event processor for contract {}",
130105
self.config.contract_addr.as_str()
131106
);
132107
break;
133108
}
134-
if let Some(number) = log.block_number {
135-
if number.as_u64() % 10 == 0 {
109+
let provider =
110+
Provider::<Ws>::connect_with_reconnects(local_evm_node_url.as_str(), 100)
111+
.await
112+
.map_err(|e| DB3Error::StoreEventError(format!("{e}")))?;
113+
let provider_arc = Arc::new(provider);
114+
let filter = match self.config.start_block == 0 {
115+
true => Filter::new().address(address),
116+
false => {
136117
info!(
137-
"contract {} sync status block {} event number {}",
138-
self.config.contract_addr.as_str(),
139-
self.block_number.load(Ordering::Relaxed),
140-
self.event_number.load(Ordering::Relaxed)
118+
"start process contract from block {} with address {}",
119+
self.config.start_block, self.config.contract_addr
141120
);
121+
Filter::new()
122+
.from_block(self.config.start_block)
123+
.address(address)
142124
}
143-
self.block_number.store(number.as_u64(), Ordering::Relaxed)
144-
}
145-
for e in abi.events() {
146-
// verify
147-
let event_signature = log
148-
.topics
149-
.get(0)
150-
.ok_or(DB3Error::StoreEventError(format!("")))?;
125+
};
151126

152-
if event_signature != &e.signature() {
153-
continue;
154-
}
155-
if !self.config.target_events.contains(e.name.as_str()) {
156-
continue;
157-
}
158-
let raw_log = RawLog {
159-
topics: log.topics.clone(),
160-
data: log.data.to_vec(),
161-
};
162-
if let Ok(log_entry) = e.parse_log(raw_log) {
163-
let json_value = Self::log_to_doc(&log_entry);
164-
match serde_json::to_string(&json_value) {
165-
Ok(value) => {
166-
let values = vec![value.to_string()];
167-
if let Err(e) = self.db_store.add_docs(
168-
&db_addr,
169-
&DB3Address::ZERO,
170-
e.name.as_str(),
171-
&values,
172-
None,
173-
) {
174-
warn!("fail to write json doc {} for {e}", value.as_str());
175-
} else {
176-
self.event_number.fetch_add(1, Ordering::Relaxed);
127+
if let Ok(mut stream) = provider_arc.clone().subscribe_logs(&filter).await {
128+
loop {
129+
if let Some(log) = stream.next().await {
130+
if !self.running.load(Ordering::Relaxed) {
131+
info!(
132+
"stop event processor for contract {}",
133+
self.config.contract_addr.as_str()
134+
);
135+
break;
136+
}
137+
if let Some(number) = log.block_number {
138+
if number.as_u64() % 10 == 0 {
139+
info!(
140+
"contract {} sync status block {} event number {}",
141+
self.config.contract_addr.as_str(),
142+
self.block_number.load(Ordering::Relaxed),
143+
self.event_number.load(Ordering::Relaxed)
144+
);
177145
}
146+
self.block_number.store(number.as_u64(), Ordering::Relaxed)
178147
}
179-
Err(e) => {
180-
warn!("fail to convert to json for {e} ");
148+
for e in abi.events() {
149+
// verify
150+
let event_signature = log
151+
.topics
152+
.get(0)
153+
.ok_or(DB3Error::StoreEventError(format!("")))?;
154+
if event_signature != &e.signature() {
155+
continue;
156+
}
157+
if !self.config.target_events.contains(e.name.as_str()) {
158+
continue;
159+
}
160+
let raw_log = RawLog {
161+
topics: log.topics.clone(),
162+
data: log.data.to_vec(),
163+
};
164+
if let Ok(log_entry) = e.parse_log(raw_log) {
165+
let json_value = Self::log_to_doc(&log_entry);
166+
match serde_json::to_string(&json_value) {
167+
Ok(value) => {
168+
let values = vec![value.to_string()];
169+
if let Err(e) = self.db_store.add_docs(
170+
&db_addr,
171+
&DB3Address::ZERO,
172+
e.name.as_str(),
173+
&values,
174+
None,
175+
) {
176+
warn!(
177+
"fail to write json doc {} for {e}",
178+
value.as_str()
179+
);
180+
} else {
181+
self.event_number.fetch_add(1, Ordering::Relaxed);
182+
}
183+
}
184+
Err(e) => {
185+
warn!("fail to convert to json for {e} ");
186+
}
187+
}
188+
break;
189+
}
181190
}
191+
} else {
192+
warn!("empty log from stream, sleep 5 seconds and reconnect to it");
193+
sleep(Duration::from_millis(5 * 1000)).await;
194+
break;
182195
}
183-
break;
184196
}
197+
} else {
198+
warn!("fail to subscribe the log, sleep 5 seconds and reconnect to it");
199+
sleep(Duration::from_millis(5 * 1000)).await;
185200
}
186201
}
202+
187203
Ok(())
188204
}
189205

‎src/event/src/meta_store_event_processor.rs‎

Lines changed: 69 additions & 66 deletions
Original file line numberDiff line numberDiff line change
@@ -347,10 +347,6 @@ impl MetaStoreEventProcessor {
347347
evm_node_url
348348
);
349349
self.last_running.store(Some(running.clone()));
350-
let provider = Provider::<Ws>::connect_with_reconnects(evm_node_url, 100)
351-
.await
352-
.map_err(|e| DB3Error::StoreEventError(format!("{e}")))?;
353-
let provider = Arc::new(provider);
354350
let address = contract_addr
355351
.parse::<Address>()
356352
.map_err(|_| DB3Error::InvalidAddress)?;
@@ -391,81 +387,88 @@ impl MetaStoreEventProcessor {
391387
}
392388
};
393389
let local_contract_addr = contract_addr.to_string();
390+
let local_evm_node_url = evm_node_url.to_string();
394391
task::spawn(async move {
395392
loop {
396393
if !running.load(Ordering::Relaxed) {
397394
break;
398395
}
399-
match provider.subscribe_logs(&filter).await {
400-
Ok(mut stream) => loop {
401-
if !running.load(Ordering::Relaxed) {
402-
info!(
403-
"stop event processor for contract {}",
404-
local_contract_addr.as_str()
405-
);
406-
break;
407-
}
408-
match stream.next().await {
409-
Some(log) => {
396+
if let Ok(provider) =
397+
Provider::<Ws>::connect_with_reconnects(local_evm_node_url.as_str(), 100).await
398+
{
399+
let provider = Arc::new(provider);
400+
match provider.subscribe_logs(&filter).await {
401+
Ok(mut stream) => loop {
402+
if !running.load(Ordering::Relaxed) {
410403
info!(
411-
"block number {:?} transacion {:?} sender address {:?} ",
412-
log.block_number, log.transaction_hash, log.address
404+
"stop event processor for contract {}",
405+
local_contract_addr.as_str()
413406
);
414-
if let (Ok(Some(transaction)), Ok(event_signature)) = (
415-
provider
416-
.get_transaction(log.transaction_hash.unwrap())
417-
.await
418-
.map_err(|e| {
419-
DB3Error::StoreEventError(format!(
420-
"fail to get transaction {e}"
421-
))
422-
}),
423-
log.topics.get(0).ok_or(DB3Error::StoreEventError(format!(
424-
"fail to get topics"
425-
))),
426-
) {
427-
if event_signature == &(CreateDatabaseFilter::signature()) {
428-
if let Ok(_) = Self::handle_create_doc_database(
429-
&log,
430-
&transaction,
431-
network,
432-
&wallet,
433-
&local_state_store,
434-
&local_storage,
435-
&local_db_store,
436-
&mutation_type,
437-
)
438-
.await
439-
{}
440-
} else if event_signature
441-
== &(CreateCollectionFilter::signature())
442-
{
443-
if let Ok(_) = Self::handle_create_collection(
444-
&log,
445-
&transaction,
446-
network,
447-
&wallet,
448-
&local_state_store,
449-
&local_storage,
450-
&local_db_store,
451-
&mutation_type,
452-
)
453-
.await
454-
{}
407+
break;
408+
}
409+
match stream.next().await {
410+
Some(log) => {
411+
info!(
412+
"block number {:?} transacion {:?} sender address {:?} ",
413+
log.block_number, log.transaction_hash, log.address
414+
);
415+
if let (Ok(Some(transaction)), Ok(event_signature)) = (
416+
provider
417+
.get_transaction(log.transaction_hash.unwrap())
418+
.await
419+
.map_err(|e| {
420+
DB3Error::StoreEventError(format!(
421+
"fail to get transaction {e}"
422+
))
423+
}),
424+
log.topics.get(0).ok_or(DB3Error::StoreEventError(
425+
format!("fail to get topics"),
426+
)),
427+
) {
428+
if event_signature == &(CreateDatabaseFilter::signature()) {
429+
if let Ok(_) = Self::handle_create_doc_database(
430+
&log,
431+
&transaction,
432+
network,
433+
&wallet,
434+
&local_state_store,
435+
&local_storage,
436+
&local_db_store,
437+
&mutation_type,
438+
)
439+
.await
440+
{}
441+
} else if event_signature
442+
== &(CreateCollectionFilter::signature())
443+
{
444+
if let Ok(_) = Self::handle_create_collection(
445+
&log,
446+
&transaction,
447+
network,
448+
&wallet,
449+
&local_state_store,
450+
&local_storage,
451+
&local_db_store,
452+
&mutation_type,
453+
)
454+
.await
455+
{}
456+
}
455457
}
456458
}
459+
None => {
460+
warn!("empty log from stream, sleep 5 seconds and reconnect to it");
461+
sleep(Duration::from_millis(5 * 1000)).await;
462+
break;
463+
}
457464
}
458-
None => {
459-
warn!("empty log from stream");
460-
sleep(Duration::from_millis(5 * 1000)).await;
461-
}
465+
},
466+
Err(e) => {
467+
warn!("fail get stream for error {e}");
462468
}
463-
},
464-
Err(e) => {
465-
warn!("fail get stream for error {e}");
466469
}
470+
sleep(Duration::from_millis(5 * 1000)).await;
467471
}
468-
sleep(Duration::from_millis(5 * 1000)).await;
469472
}
470473
warn!("the meta contract event listener exits");
471474
});

0 commit comments

Comments
 (0)