fix(pool): fix pool max ops per sender during replacement

crates/pool/src/mempool/reputation.rs

@@ -120,8 +120,8 @@ impl AddressReputation {
         self.state.write().add_seen(address);
     }
 
-    pub(crate) fn remove_seen(&self, address: Address, value: u64) {
-        self.state.write().remove_seen(address, value);
+    pub(crate) fn dec_seen(&self, address: Address) {
+        self.state.write().dec_seen(address);
     }
 
     pub(crate) fn handle_urep_030_penalty(&self, address: Address) {
@@ -132,6 +132,10 @@ impl AddressReputation {
         self.state.write().handle_srep_050_penalty(address);
     }
 
+    pub(crate) fn handle_erep_015_amendment(&self, address: Address, value: u64) {
+        self.state.write().handle_erep_015_amendment(address, value);
+    }
+
     pub(crate) fn dump_reputation(&self) -> Vec<Reputation> {
         self.state.read().dump_reputation()
     }
@@ -140,8 +144,8 @@ impl AddressReputation {
         self.state.write().add_included(address);
     }
 
-    pub(crate) fn remove_included(&self, address: Address) {
-        self.state.write().remove_included(address);
+    pub(crate) fn dec_included(&self, address: Address) {
+        self.state.write().dec_included(address);
     }
 
     pub(crate) fn set_reputation(&self, address: Address, ops_seen: u64, ops_included: u64) {
@@ -222,9 +226,9 @@ impl AddressReputationInner {
         count.ops_seen += 1;
     }
 
-    fn remove_seen(&mut self, address: Address, value: u64) {
+    fn dec_seen(&mut self, address: Address) {
         let count = self.counts.entry(address).or_default();
-        count.ops_seen = count.ops_seen.saturating_sub(value);
+        count.ops_seen = count.ops_seen.saturating_sub(1);
     }
 
     fn handle_urep_030_penalty(&mut self, address: Address) {
@@ -238,6 +242,11 @@ impl AddressReputationInner {
         count.ops_seen = self.params.bundle_invalidation_ops_seen_staked_penalty;
     }
 
+    pub(crate) fn handle_erep_015_amendment(&mut self, address: Address, value: u64) {
+        let count = self.counts.entry(address).or_default();
+        count.ops_seen = count.ops_seen.saturating_sub(value);
+    }
+
     fn dump_reputation(&self) -> Vec<Reputation> {
         self.counts
             .iter()
@@ -254,7 +263,7 @@ impl AddressReputationInner {
         count.ops_included += 1;
     }
 
-    fn remove_included(&mut self, address: Address) {
+    fn dec_included(&mut self, address: Address) {
         let count = self.counts.entry(address).or_default();
         count.ops_included = count.ops_included.saturating_sub(1)
     }
@@ -325,8 +334,8 @@ mod tests {
         assert_eq!(counts.ops_seen, 1000);
         assert_eq!(counts.ops_included, 1000);
 
-        reputation.remove_seen(addr, 1);
-        reputation.remove_included(addr);
+        reputation.dec_seen(addr);
+        reputation.dec_included(addr);
         let counts = reputation.counts.get(&addr).unwrap();
         assert_eq!(counts.ops_seen, 999);
         assert_eq!(counts.ops_included, 999);

crates/pool/src/mempool/uo_pool.rs

@@ -292,7 +292,7 @@ where
 
             if let Some(po) = pool_op {
                 for entity_addr in po.entities().map(|e| e.address).unique() {
-                    self.reputation.remove_included(entity_addr);
+                    self.reputation.dec_included(entity_addr);
                 }
 
                 unmined_op_count += 1;
@@ -468,9 +468,6 @@ where
         origin: OperationOrigin,
         op: UserOperationVariant,
     ) -> MempoolResult<B256> {
-        // TODO(danc) aggregator reputation is not implemented
-        // TODO(danc) catch ops with aggregators prior to simulation and reject
-
         // Check reputation of entities in involved in the operation
         // If throttled, entity can have THROTTLED_ENTITY_MEMPOOL_COUNT inflight operation at a time, else reject
         // If banned, reject
@@ -521,6 +518,8 @@ where
         // Check if op is already known or replacing another, and if so, ensure its fees are high enough
         // do this before simulation to save resources
         let replacement = self.state.read().pool.check_replacement(&op)?;
+        let to_replace = replacement.and_then(|r| self.state.read().pool.get_operation_by_hash(r));
+
         // Check if op violates the STO-040 spec rule
         self.state.read().pool.check_multiple_roles_violation(&op)?;
 
@@ -587,6 +586,7 @@ where
         {
             let state = self.state.read();
             if !pool_op.account_is_staked
+                && to_replace.is_none()
                 && state.pool.address_count(&pool_op.uo.sender())
                     >= self.config.same_sender_mempool_count
             {
@@ -599,9 +599,16 @@ where
             // Check unstaked non-sender entity counts in the mempool
             for entity in pool_op
                 .unstaked_entities()
+                .unique()
                 .filter(|e| e.address != pool_op.entity_infos.sender.address())
             {
-                let ops_allowed = self.reputation.get_ops_allowed(entity.address);
+                let mut ops_allowed = self.reputation.get_ops_allowed(entity.address);
+                if let Some(to_replace) = &to_replace {
+                    if to_replace.entities().contains(&entity) {
+                        ops_allowed += 1;
+                    }
+                }
+
                 if state.pool.address_count(&entity.address) >= ops_allowed as usize {
                     return Err(MempoolError::MaxOperationsReached(
                         ops_allowed as usize,
@@ -628,17 +635,20 @@ where
         // once the operation has been added to the pool
         self.paymaster.add_or_update_balance(&pool_op).await?;
 
-        // Update reputation
-        if replacement.is_none() {
-            pool_op.entities().unique().for_each(|e| {
-                self.reputation.add_seen(e.address);
-                if self.reputation.status(e.address) == ReputationStatus::Throttled {
-                    self.throttle_entity(e);
-                } else if self.reputation.status(e.address) == ReputationStatus::Banned {
-                    self.remove_entity(e);
-                }
+        // Update reputation, handling replacement if needed
+        if let Some(to_replace) = to_replace {
+            to_replace.entities().unique().for_each(|e| {
+                self.reputation.dec_seen(e.address);
             });
         }
+        pool_op.entities().unique().for_each(|e| {
+            self.reputation.add_seen(e.address);
+            if self.reputation.status(e.address) == ReputationStatus::Throttled {
+                self.throttle_entity(e);
+            } else if self.reputation.status(e.address) == ReputationStatus::Banned {
+                self.remove_entity(e);
+            }
+        });
 
         // Emit event
         let op_hash = pool_op
@@ -735,17 +745,19 @@ where
                     entity.is_paymaster(),
                     "Attempted to add EREP-015 paymaster amendment for non-paymaster entity"
                 );
-                assert!(
-                    update.value.is_some(),
-                    "PaymasterOpsSeenDecrement must carry an explicit decrement value"
+                self.reputation.handle_erep_015_amendment(
+                    entity.address,
+                    update
+                        .value
+                        .expect("PaymasterOpsSeenDecrement must carry an explicit decrement value"),
                 );
-                self.reputation
-                    .remove_seen(entity.address, update.value.unwrap());
             }
         }
 
         if self.reputation.status(entity.address) == ReputationStatus::Banned {
             self.remove_entity(entity);
+        } else if self.reputation.status(entity.address) == ReputationStatus::Throttled {
+            self.throttle_entity(entity);
         }
     }