Skip to content

Commit 678535f

Browse files
sameh-farouksameh-farouk
committed
fix cargo fmt
1 parent facd559 commit 678535f

1 file changed

Lines changed: 74 additions & 39 deletions

File tree

‎src/events/mod.rs‎

Lines changed: 74 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -3,41 +3,42 @@ use std::{collections::LinkedList, time::Duration};
33
use crate::{cache::Cache, tfchain::tfchain, twin::Twin};
44
use anyhow::Result;
55
use futures::StreamExt;
6+
use lazy_static::lazy_static;
67
use log;
7-
use subxt::{config::Header, OnlineClient, PolkadotConfig, blocks::Block as SubxtBlock};
88
use prometheus::{IntCounter, IntCounterVec, IntGauge, Opts, Registry};
9-
use lazy_static::lazy_static;
9+
use subxt::{blocks::Block as SubxtBlock, config::Header, OnlineClient, PolkadotConfig};
1010

1111
lazy_static! {
1212
static ref EVENTS_RECONNECTING: IntGauge = IntGauge::new(
1313
"events_listener_reconnecting",
1414
"1 while reconnecting/backing off; 0 otherwise",
15-
).unwrap();
16-
15+
)
16+
.unwrap();
1717
static ref EVENTS_RECONNECT_CYCLES: IntCounter = IntCounter::new(
1818
"events_listener_reconnect_cycles_total",
1919
"Successful reconnect cycles",
20-
).unwrap();
21-
20+
)
21+
.unwrap();
2222
static ref EVENTS_ERRORS: IntCounterVec = IntCounterVec::new(
2323
Opts::new("events_listener_errors_total", "Errors by stage"),
2424
&["stage"],
25-
).unwrap();
26-
25+
)
26+
.unwrap();
2727
static ref EVENTS_LAST_BLOCK_NUM: IntGauge = IntGauge::new(
2828
"events_listener_last_processed_block_number",
2929
"Last best-head block number processed",
30-
).unwrap();
31-
30+
)
31+
.unwrap();
3232
static ref EVENTS_TWIN_STORED_TOTAL: IntCounter = IntCounter::new(
3333
"events_listener_twin_stored_total",
3434
"Total number of TwinStored events processed",
35-
).unwrap();
36-
35+
)
36+
.unwrap();
3737
static ref EVENTS_TWIN_UPDATED_TOTAL: IntCounter = IntCounter::new(
3838
"events_listener_twin_updated_total",
3939
"Total number of TwinUpdated events processed",
40-
).unwrap();
40+
)
41+
.unwrap();
4142
}
4243

4344
pub struct EventListenerOptions {
@@ -74,12 +75,22 @@ impl EventListenerOptions {
7475
cache.flush().await?;
7576

7677
// Register metrics into provided registry (ignore duplicate registration errors)
77-
let _ = self.registry.register(Box::new(EVENTS_RECONNECTING.clone()));
78-
let _ = self.registry.register(Box::new(EVENTS_RECONNECT_CYCLES.clone()));
78+
let _ = self
79+
.registry
80+
.register(Box::new(EVENTS_RECONNECTING.clone()));
81+
let _ = self
82+
.registry
83+
.register(Box::new(EVENTS_RECONNECT_CYCLES.clone()));
7984
let _ = self.registry.register(Box::new(EVENTS_ERRORS.clone()));
80-
let _ = self.registry.register(Box::new(EVENTS_LAST_BLOCK_NUM.clone()));
81-
let _ = self.registry.register(Box::new(EVENTS_TWIN_STORED_TOTAL.clone()));
82-
let _ = self.registry.register(Box::new(EVENTS_TWIN_UPDATED_TOTAL.clone()));
85+
let _ = self
86+
.registry
87+
.register(Box::new(EVENTS_LAST_BLOCK_NUM.clone()));
88+
let _ = self
89+
.registry
90+
.register(Box::new(EVENTS_TWIN_STORED_TOTAL.clone()));
91+
let _ = self
92+
.registry
93+
.register(Box::new(EVENTS_TWIN_UPDATED_TOTAL.clone()));
8394

8495
Ok(Listener {
8596
api,
@@ -109,7 +120,9 @@ where
109120
{
110121
pub async fn new(substrate_urls: Vec<String>, cache: C) -> Result<Self> {
111122
// Delegate to options with defaults (threshold=600, default registry)
112-
EventListenerOptions::new().build(substrate_urls, cache).await
123+
EventListenerOptions::new()
124+
.build(substrate_urls, cache)
125+
.await
113126
}
114127

115128
async fn connect(urls: &mut LinkedList<String>) -> Result<OnlineClient<PolkadotConfig>> {
@@ -155,7 +168,12 @@ where
155168
if head_num > last {
156169
let gap = head_num - last;
157170
if gap <= self.catchup_threshold {
158-
log::info!("catching up {} blocks ({} -> {}) without flushing cache", gap, last, head_num);
171+
log::info!(
172+
"catching up {} blocks ({} -> {}) without flushing cache",
173+
gap,
174+
last,
175+
head_num
176+
);
159177
// Sequential catch-up; yield periodically
160178
let mut count = 0u64;
161179
for n in (last + 1)..=head_num {
@@ -164,11 +182,18 @@ where
164182
self.process_block(block).await?;
165183
}
166184
count += 1;
167-
if count % 50 == 0 { tokio::task::yield_now().await; }
185+
if count % 50 == 0 {
186+
tokio::task::yield_now().await;
187+
}
168188
}
169189
Ok(())
170190
} else {
171-
log::warn!("gap {} exceeds threshold {}; flushing cache and jumping to head {}", gap, self.catchup_threshold, head_num);
191+
log::warn!(
192+
"gap {} exceeds threshold {}; flushing cache and jumping to head {}",
193+
gap,
194+
self.catchup_threshold,
195+
head_num
196+
);
172197
self.cache.flush().await?;
173198
self.last_processed = Some(head_num);
174199
Ok(())
@@ -214,15 +239,16 @@ where
214239

215240
async fn handle_events_inner(&mut self) -> Result<()> {
216241
// Subscribe to best head to minimize latency; reorg safety can be layered later if needed.
217-
let mut blocks_sub = self.api.blocks().subscribe_best().await.map_err(|e| {
218-
e
219-
})?;
242+
let mut blocks_sub = self.api.blocks().subscribe_best().await.map_err(|e| e)?;
220243
// Detect stalled streams: if no new block arrives within this window, bail to trigger reconnect.
221244
let stall_timeout = Duration::from_secs(7);
222245
while let Some(block) = match tokio::time::timeout(stall_timeout, blocks_sub.next()).await {
223246
Ok(v) => v,
224247
Err(_) => {
225-
return Err(anyhow::anyhow!("block subscription timed out after {:?}", stall_timeout));
248+
return Err(anyhow::anyhow!(
249+
"block subscription timed out after {:?}",
250+
stall_timeout
251+
));
226252
}
227253
} {
228254
let block = block?;
@@ -232,15 +258,16 @@ where
232258
}
233259

234260
// Unified per-block processing: updates metrics, last_processed, and handles events
235-
async fn process_block(&mut self, block: SubxtBlock<PolkadotConfig, OnlineClient<PolkadotConfig>>) -> Result<()> {
261+
async fn process_block(
262+
&mut self,
263+
block: SubxtBlock<PolkadotConfig, OnlineClient<PolkadotConfig>>,
264+
) -> Result<()> {
236265
let header = block.header();
237266
let num_i64: i64 = header.number.into();
238267
EVENTS_LAST_BLOCK_NUM.set(num_i64);
239268
log::trace!("processing block number: {}", num_i64);
240269

241-
let events = block.events().await.map_err(|e| {
242-
e
243-
})?;
270+
let events = block.events().await.map_err(|e| e)?;
244271
for evt in events.iter() {
245272
let evt = match evt {
246273
Err(err) => {
@@ -251,16 +278,24 @@ where
251278
Ok(e) => e,
252279
};
253280
if let Ok(Some(twin)) = evt.as_event::<tfchain::tfgrid_module::events::TwinStored>() {
254-
self.cache.set(twin.0.id, twin.0.into()).await.map_err(|e| {
255-
EVENTS_ERRORS.with_label_values(&["cache_set"]).inc();
256-
e
257-
})?;
281+
self.cache
282+
.set(twin.0.id, twin.0.into())
283+
.await
284+
.map_err(|e| {
285+
EVENTS_ERRORS.with_label_values(&["cache_set"]).inc();
286+
e
287+
})?;
258288
EVENTS_TWIN_STORED_TOTAL.inc();
259-
} else if let Ok(Some(twin)) = evt.as_event::<tfchain::tfgrid_module::events::TwinUpdated>() {
260-
self.cache.set(twin.0.id, twin.0.into()).await.map_err(|e| {
261-
EVENTS_ERRORS.with_label_values(&["cache_set"]).inc();
262-
e
263-
})?;
289+
} else if let Ok(Some(twin)) =
290+
evt.as_event::<tfchain::tfgrid_module::events::TwinUpdated>()
291+
{
292+
self.cache
293+
.set(twin.0.id, twin.0.into())
294+
.await
295+
.map_err(|e| {
296+
EVENTS_ERRORS.with_label_values(&["cache_set"]).inc();
297+
e
298+
})?;
264299
EVENTS_TWIN_UPDATED_TOTAL.inc();
265300
}
266301
}

0 commit comments

Comments
 (0)