Added limit for IWANT requests

0xMimir · directcuteo · commit 45ae89bc168a · 2024-10-23T11:40:49.000+03:00
Small improvements
diff --git a/frontend/src/app/core/services/web-node.service.ts b/frontend/src/app/core/services/web-node.service.ts
@@ -1,16 +1,15 @@
 import { Injectable } from '@angular/core';
 import { BehaviorSubject, filter, from, fromEvent, map, merge, Observable, of, switchMap, tap } from 'rxjs';
 import base from 'base-x';
-import { any, log } from '@openmina/shared';
+import { any } from '@openmina/shared';
 import { HttpClient } from '@angular/common/http';
 
 @Injectable({
   providedIn: 'root',
 })
 export class WebNodeService {
 
-  private readonly backendSubject$: BehaviorSubject<any> = new BehaviorSubject<any>(null);
-  private backend: any;
+  private readonly webnode$: BehaviorSubject<any> = new BehaviorSubject<any>(null);
   private webNodeKeyPair: { publicKey: string, privateKey: string };
 
   constructor(private http: HttpClient) {
@@ -42,79 +41,74 @@ export class WebNodeService {
           console.log(wasm);
           return from(wasm.run(this.webNodeKeyPair.privateKey));
         }),
-        tap((jsHandle: any) => {
-          this.backend = jsHandle;
+        tap((webnode: any) => {
           console.log('----------------WEBNODE----------------');
-          console.log(jsHandle);
-          this.backendSubject$.next(jsHandle);
+          console.log(webnode);
+          this.webnode$.next(webnode);
         }),
-        switchMap(() => this.backendSubject$.asObservable()),
+        switchMap(() => this.webnode$.asObservable()),
         filter(Boolean),
       );
   }
 
-  get webNodeKeys(): { publicKey: string, privateKey: string } {
-    return this.webNodeKeyPair;
-  }
-
   get status$(): Observable<any> {
-    return this.backendSubject$.asObservable().pipe(
+    return this.webnode$.asObservable().pipe(
       filter(Boolean),
       switchMap(handle => from((handle as any).status())),
     );
   }
 
   get blockProducerStats$(): Observable<any> {
-    return this.backendSubject$.asObservable().pipe(
+    return this.webnode$.asObservable().pipe(
       filter(Boolean),
       switchMap(handle => from((handle as any).stats().block_producer())),
     );
   }
 
   get peers$(): Observable<any> {
-    return this.backendSubject$.asObservable().pipe(
+    return this.webnode$.asObservable().pipe(
       filter(Boolean),
       switchMap(handle => from(any(handle).state().peers())),
     );
   }
 
   get messageProgress$(): Observable<any> {
-    return this.backendSubject$.asObservable().pipe(
+    return this.webnode$.asObservable().pipe(
       filter(Boolean),
       switchMap(handle => from((handle as any).state().message_progress())),
     );
   }
 
   get sync$(): Observable<any> {
-    return this.backendSubject$.asObservable().pipe(
+    return this.webnode$.asObservable().pipe(
       filter(Boolean),
       switchMap(handle => from((handle as any).stats().sync())),
     );
   }
 
   get accounts$(): Observable<any> {
-    return this.backendSubject$.asObservable().pipe(
+    return this.webnode$.asObservable().pipe(
       filter(Boolean),
       switchMap(handle => from((handle as any).ledger().latest().accounts().all())),
     );
   }
 
   get bestChainUserCommands$(): Observable<any> {
-    return this.backendSubject$.asObservable().pipe(
+    return this.webnode$.asObservable().pipe(
       filter(Boolean),
       switchMap(handle => from((handle as any).transition_frontier().best_chain().user_commands())),
     );
   }
 
   sendPayment$(payment: any): Observable<any> {
-    return this.backendSubject$.asObservable().pipe(
+    return this.webnode$.asObservable().pipe(
       filter(Boolean),
       switchMap(handle => from((handle as any).transaction_pool().inject().payment(payment))),
     );
   }
 
   get transactionPool$(): Observable<any> {
-    return this.backendSubject$.asObservable().pipe(
+    return this.webnode$.asObservable().pipe(
       filter(Boolean),
       switchMap(handle => from((handle as any).transaction_pool().get())),
     );
diff --git a/frontend/src/app/features/benchmarks/wallets/benchmarks-wallets.effects.ts b/frontend/src/app/features/benchmarks/wallets/benchmarks-wallets.effects.ts
@@ -2,16 +2,17 @@ import { Injectable } from '@angular/core';
 import { MinaState, selectMinaState } from '@app/app.setup';
 import { Actions, createEffect, ofType } from '@ngrx/effects';
 import { Effect } from '@openmina/shared';
-import { forkJoin, map, switchMap } from 'rxjs';
+import { EMPTY, forkJoin, map, switchMap } from 'rxjs';
 import { Store } from '@ngrx/store';
 import {
+  BENCHMARKS_WALLETS_CLOSE,
   BENCHMARKS_WALLETS_GET_ALL_TXS,
   BENCHMARKS_WALLETS_GET_ALL_TXS_SUCCESS,
   BENCHMARKS_WALLETS_GET_WALLETS,
   BENCHMARKS_WALLETS_GET_WALLETS_SUCCESS,
   BENCHMARKS_WALLETS_SEND_TX_SUCCESS,
   BENCHMARKS_WALLETS_SEND_TXS, BENCHMARKS_WALLETS_SEND_ZKAPPS, BENCHMARKS_WALLETS_SEND_ZKAPPS_SUCCESS,
-  BenchmarksWalletsActions,
+  BenchmarksWalletsActions, BenchmarksWalletsClose,
   BenchmarksWalletsGetWallets,
   BenchmarksWalletsSendTxs,
 } from '@benchmarks/wallets/benchmarks-wallets.actions';
@@ -38,22 +39,23 @@ export class BenchmarksWalletsEffects extends MinaRustBaseEffect<BenchmarksWalle
               private zkService: BenchmarksWalletsZkService,
               private mempoolService: MempoolService,
               store: Store<MinaState>) {
-
     super(store, selectMinaState);
 
     this.getWallets$ = createEffect(() => this.actions$.pipe(
-      ofType(BENCHMARKS_WALLETS_GET_WALLETS),
-      this.latestActionState<BenchmarksWalletsGetWallets>(),
-      switchMap(({ action }) => this.benchmarksService.getAccounts().pipe(
-        switchMap(payload => {
-          const actions = [];
-          if (action.payload?.initialRequest) {
-            actions.push({ type: BENCHMARKS_WALLETS_GET_ALL_TXS });
-          }
-          actions.push({ type: BENCHMARKS_WALLETS_GET_WALLETS_SUCCESS, payload });
-          return actions;
-        }),
-      )),
+      ofType(BENCHMARKS_WALLETS_GET_WALLETS, BENCHMARKS_WALLETS_CLOSE),
+      this.latestActionState<BenchmarksWalletsGetWallets | BenchmarksWalletsClose>(),
+      switchMap(({ action }) => action.type === BENCHMARKS_WALLETS_CLOSE
+        ? EMPTY
+        : this.benchmarksService.getAccounts().pipe(
+          switchMap(payload => {
+            const actions = [];
+            if (action.payload?.initialRequest) {
+              actions.push({ type: BENCHMARKS_WALLETS_GET_ALL_TXS });
+            }
+            actions.push({ type: BENCHMARKS_WALLETS_GET_WALLETS_SUCCESS, payload });
+            return actions;
+          }),
+        )),
       catchErrorAndRepeat(MinaErrorType.GENERIC, BENCHMARKS_WALLETS_GET_WALLETS_SUCCESS, []),
     ));
 
diff --git a/frontend/src/app/features/benchmarks/wallets/benchmarks-wallets.service.ts b/frontend/src/app/features/benchmarks/wallets/benchmarks-wallets.service.ts
@@ -4035,7 +4035,7 @@ export class BenchmarksWalletsService {
   }
 
   getAccounts(): Observable<Pick<BenchmarksWallet, 'publicKey' | 'privateKey' | 'minaTokens' | 'nonce'>[]> {
-    return this.rust.get<any[]>('/accounts').pipe(
+    return this.rust.get<AccountsResponse[]>('/accounts').pipe(
       map(wallets => wallets.map(wallet => {
         return ({
           privateKey: WALLETS.find(w => w.publicKey === wallet.public_key)?.privateKey,
@@ -4140,3 +4140,9 @@ export class BenchmarksWalletsService {
       });
   }
 }
+
+export interface AccountsResponse {
+  public_key: string;
+  nonce: number;
+  balance: number;
+}
diff --git a/frontend/src/app/layout/server-status/server-status.component.html b/frontend/src/app/layout/server-status/server-status.component.html
@@ -24,7 +24,7 @@
     </div>
   </ng-container>
   <div class="node-status fx-row-vert-cent bg-surface h-sm border-rad-6 p-relative z-1 mr-10"
-       [class.can-add-nodes]="canAddNodes">
+       [class.can-add-nodes]="switchForbidden">
     <ng-container *ngIf="!switchForbidden && !hideNodeStats">
       <div class="shine-parent overflow-hidden p-absolute z-0 border-rad-6">
         <div *ngIf="details.status === AppNodeStatus.CATCHUP || details.status === AppNodeStatus.BOOTSTRAP"
diff --git a/frontend/src/app/layout/server-status/server-status.component.scss b/frontend/src/app/layout/server-status/server-status.component.scss
@@ -113,6 +113,11 @@
     }
   }
 
+  .chip::before {
+    border-top-right-radius: 0 !important;
+    border-bottom-right-radius: 0 !important;
+  }
+
   .shine-parent {
     height: calc(100% + 3px);
     width: calc(100% - 2px);
diff --git a/frontend/src/app/layout/toolbar/loading.reducer.ts b/frontend/src/app/layout/toolbar/loading.reducer.ts
@@ -45,6 +45,7 @@ import {
 } from '@network/bootstrap-stats/network-bootstrap-stats.actions';
 import { BLOCK_PRODUCTION_PREFIX } from '@block-production/block-production.actions';
 import {
+  BENCHMARKS_WALLETS_CLOSE,
   BENCHMARKS_WALLETS_GET_ALL_TXS,
   BENCHMARKS_WALLETS_GET_ALL_TXS_SUCCESS,
   BENCHMARKS_WALLETS_GET_WALLETS, BENCHMARKS_WALLETS_GET_WALLETS_SUCCESS,
@@ -154,6 +155,8 @@ export function loadingReducer(state: LoadingState = initialState, action: Featu
       return remove(state, BENCHMARKS_WALLETS_GET_WALLETS);
     case BENCHMARKS_WALLETS_GET_ALL_TXS_SUCCESS:
       return remove(state, BENCHMARKS_WALLETS_GET_ALL_TXS);
+    case BENCHMARKS_WALLETS_CLOSE:
+      return remove(state, [BENCHMARKS_WALLETS_GET_WALLETS, BENCHMARKS_WALLETS_GET_ALL_TXS]);
 
     default:
       return state;
diff --git a/p2p/src/network/pubsub/p2p_network_pubsub_reducer.rs b/p2p/src/network/pubsub/p2p_network_pubsub_reducer.rs
@@ -3,7 +3,7 @@ use std::{collections::btree_map::Entry, sync::Arc};
 use binprot::BinProtRead;
 use mina_p2p_messages::{gossip, v2};
 use openmina_core::{block::BlockWithHash, bug_condition, fuzz_maybe, fuzzed_maybe, Substate};
-use redux::Dispatcher;
+use redux::{Dispatcher, Timestamp};
 
 use crate::{
     channels::{snark::P2pChannelsSnarkAction, transaction::P2pChannelsTransactionAction},
@@ -28,7 +28,7 @@ impl P2pNetworkPubsubState {
         Action: crate::P2pActionTrait<State>,
     {
         let pubsub_state = state_context.get_substate_mut()?;
-        let (action, _meta) = action.split();
+        let (action, meta) = action.split();
 
         match action {
             P2pNetworkPubsubAction::NewStream {
@@ -125,7 +125,7 @@ impl P2pNetworkPubsubState {
                 seen_limit,
                 ..
             } => {
-                pubsub_state.reduce_incoming_data(&peer_id, data)?;
+                pubsub_state.reduce_incoming_data(&peer_id, data, meta.time())?;
 
                 let dispatcher: &mut Dispatcher<Action, State> = state_context.into_dispatcher();
                 dispatcher.push(P2pNetworkPubsubEffectfulAction::IncomingData {
@@ -447,7 +447,12 @@ impl P2pNetworkPubsubState {
         Ok(())
     }
 
-    fn reduce_incoming_data(&mut self, peer_id: &PeerId, data: Data) -> Result<(), String> {
+    fn reduce_incoming_data(
+        &mut self,
+        peer_id: &PeerId,
+        data: Data,
+        timestamp: Timestamp,
+    ) -> Result<(), String> {
         let Some(state) = self.clients.get_mut(peer_id) else {
             // TODO: investigate, cannot reproduce this
             // bug_condition!("State not found for action: P2pNetworkPubsubAction::IncomingData");
@@ -529,14 +534,19 @@ impl P2pNetworkPubsubState {
             }
         }
 
-        for ihave in &control.ihave {
-            let message_ids = ihave
-                .message_ids
-                .iter()
-                .filter(|msg_id| !self.mcache.map.contains_key(*msg_id))
-                .cloned()
-                .collect();
-            if let Some(client) = self.clients.get_mut(peer_id) {
+        for ihave in control.ihave {
+            if self.clients.contains_key(peer_id) {
+                let message_ids = ihave
+                    .message_ids
+                    .into_iter()
+                    .filter(|message_id| self.filter_iwant_message_ids(message_id, timestamp))
+                    .collect::<Vec<_>>();
+
+                let Some(client) = self.clients.get_mut(peer_id) else {
+                    bug_condition!("State not found for {}", peer_id);
+                    return Ok(());
+                };
+
                 let ctr = client.message.control.get_or_insert_with(Default::default);
                 ctr.iwant.push(pb::ControlIWant { message_ids })
             }
diff --git a/p2p/src/network/pubsub/p2p_network_pubsub_state.rs b/p2p/src/network/pubsub/p2p_network_pubsub_state.rs
@@ -1,12 +1,18 @@
 use super::pb;
 use crate::{token::BroadcastAlgorithm, ConnectionAddr, PeerId, StreamId};
 
-use std::collections::{BTreeMap, VecDeque};
+use std::{
+    collections::{BTreeMap, VecDeque},
+    time::Duration,
+};
 
 use mina_p2p_messages::v2;
 use openmina_core::{snark::Snark, transaction::Transaction};
+use redux::Timestamp;
 use serde::{Deserialize, Serialize};
 
+pub const IWANT_TIMEOUT_DURATION: Duration = Duration::from_secs(5);
+
 #[derive(Default, Serialize, Deserialize, Debug, Clone)]
 pub struct P2pNetworkPubsubState {
     pub clients: BTreeMap<PeerId, P2pNetworkPubsubClientState>,
@@ -18,12 +24,65 @@ pub struct P2pNetworkPubsubState {
     pub incoming_transactions: Vec<(Transaction, u32)>,
     pub incoming_snarks: Vec<(Snark, u32)>,
     pub topics: BTreeMap<String, BTreeMap<PeerId, P2pNetworkPubsubClientTopicState>>,
+    pub iwant: VecDeque<P2pNetworkPubsubIwantRequestCount>,
+}
+
+#[derive(Default, Serialize, Deserialize, Debug, Clone)]
+pub struct P2pNetworkPubsubIwantRequestCount {
+    pub message_id: Vec<u8>,
+    pub count: Vec<Timestamp>,
 }
 
 impl P2pNetworkPubsubState {
     pub fn prune_peer_state(&mut self, peer_id: &PeerId) {
         self.clients.remove(peer_id);
     }
+
+    pub fn filter_iwant_message_ids(&mut self, message_id: &Vec<u8>, timestamp: Timestamp) -> bool {
+        if self.mcache.map.contains_key(message_id) {
+            return false;
+        }
+
+        let message_count = self
+            .iwant
+            .iter_mut()
+            .find(|message| &message.message_id == message_id);
+
+        match message_count {
+            Some(message) => {
+                let message_counts = std::mem::take(&mut message.count);
+
+                message.count = message_counts
+                    .into_iter()
+                    .filter(|time| {
+                        timestamp
+                            .checked_sub(*time)
+                            .map_or(false, |duration| duration < IWANT_TIMEOUT_DURATION)
+                    })
+                    .collect();
+
+                if message.count.len() < 3 {
+                    message.count.push(timestamp);
+                    return true;
+                }
+
+                false
+            }
+            None => {
+                let message_count = P2pNetworkPubsubIwantRequestCount {
+                    message_id: message_id.to_owned(),
+                    count: vec![timestamp],
+                };
+
+                self.iwant.push_back(message_count);
+                if self.iwant.len() > 10 {
+                    self.iwant.pop_front();
+                }
+
+                true
+            }
+        }
+    }
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone)]

Original file line number	Diff line number	Diff line change
`@@ -4035,7 +4035,7 @@ export class BenchmarksWalletsService {`
`4035`	`4035`	`}`
`4036`	`4036`
`4037`	`4037`	`getAccounts(): Observable<Pick<BenchmarksWallet, 'publicKey' \| 'privateKey' \| 'minaTokens' \| 'nonce'>[]> {`
`4038`		`- return this.rust.get<any[]>('/accounts').pipe(`
	`4038`	`+ return this.rust.get<AccountsResponse[]>('/accounts').pipe(`
`4039`	`4039`	`map(wallets => wallets.map(wallet => {`
`4040`	`4040`	`return ({`
`4041`	`4041`	`privateKey: WALLETS.find(w => w.publicKey === wallet.public_key)?.privateKey,`
`@@ -4140,3 +4140,9 @@ export class BenchmarksWalletsService {`
`4140`	`4140`	`});`
`4141`	`4141`	`}`
`4142`	`4142`	`}`
	`4143`	`+`
	`4144`	`+export interface AccountsResponse {`
	`4145`	`+ public_key: string;`
	`4146`	`+ nonce: number;`
	`4147`	`+ balance: number;`
	`4148`	`+}`
Original file line number	Diff line number	Diff line change
`@@ -113,6 +113,11 @@`
`113`	`113`	`}`
`114`	`114`	`}`
`115`	`115`
	`116`	`+ .chip::before {`
	`117`	`+ border-top-right-radius: 0 !important;`
	`118`	`+ border-bottom-right-radius: 0 !important;`
	`119`	`+ }`
	`120`	`+`
`116`	`121`	`.shine-parent {`
`117`	`122`	`height: calc(100% + 3px);`
`118`	`123`	`width: calc(100% - 2px);`