diff --git a/Cargo.lock b/Cargo.lock index 7e5269b1d06..7a790bd4eda 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6078,7 +6078,6 @@ dependencies = [ "parity-scale-codec", "scopeguard", "sp-core", - "thiserror 2.0.17", "tokio", "tower 0.4.13", "tower-http 0.5.2", diff --git a/ethexe/rpc/server/Cargo.toml b/ethexe/rpc/server/Cargo.toml index 4566499801d..22a462c9fcb 100644 --- a/ethexe/rpc/server/Cargo.toml +++ b/ethexe/rpc/server/Cargo.toml @@ -34,7 +34,6 @@ dashmap.workspace = true metrics.workspace = true metrics-derive.workspace = true gear-workspace-hack.workspace = true -thiserror.workspace = true scopeguard.workspace = true moka = { workspace = true, features = ["sync"] } diff --git a/ethexe/rpc/server/src/apis/injected/mod.rs b/ethexe/rpc/server/src/apis/injected/mod.rs index 0ba49ad0c08..40e6e7cb440 100644 --- a/ethexe/rpc/server/src/apis/injected/mod.rs +++ b/ethexe/rpc/server/src/apis/injected/mod.rs @@ -14,19 +14,22 @@ //! purged transaction it contains [`PurgedTransaction`](ethexe_common::injected::PurgedTransaction). //! //! [`promise_manager::PromiseSubscriptionManager`] owns the RPC-side joining logic. It keeps: -//! - one-shot subscribers keyed by transaction hash; +//! - one `watch` channel per transaction hash, fanning the receipt out to any number of +//! concurrent watchers (subscribers are anonymous receiver clones, no per-subscriber id); //! - full promises already computed locally and stored in the database; //! - compact promise receipts whose full promise body has not been observed yet. //! //! ### Subscription Setup //! //! [`InjectedApi::send_transaction_and_watch`](server::InjectedApi::send_transaction_and_watch) -//! first registers a subscriber for the transaction hash, then relays the transaction. If relaying -//! fails or the transaction is rejected before it enters the injected transaction pool, the -//! registration is cancelled and the subscription request fails. If the transaction is accepted, -//! [`spawner::spawn_pending_subscriber`] waits for a single -//! [`SignedTxReceipt`](ethexe_common::injected::SignedTxReceipt) and forwards it to the JSON-RPC -//! subscription sink. +//! first checks whether a receipt is already stored for the transaction hash. If so, it accepts +//! the subscription and delivers the cached result immediately without relaying (the `Ready` path). +//! Otherwise it registers a new pending subscriber and relays the transaction; the relayer shares +//! a single in-flight relay per transaction hash, so concurrent watchers (and plain +//! `injected_sendTransaction` callers) of one transaction produce one relay and observe the same +//! Accept/Reject outcome. Once accepted, [`spawner::spawn_pending_subscriber`] waits for the +//! receipt, which is fanned out to every registered watcher; a late watcher whose receipt is +//! already stored is served immediately from the database. //! //! **Important:** the pending subscriber is dropped after **20 * Ethereum slot** seconds to avoid //! dead subscribers. A later receipt can still be stored in the database and returned by diff --git a/ethexe/rpc/server/src/apis/injected/promise_manager.rs b/ethexe/rpc/server/src/apis/injected/promise_manager.rs index a2dcbeedb71..cb010502e12 100644 --- a/ethexe/rpc/server/src/apis/injected/promise_manager.rs +++ b/ethexe/rpc/server/src/apis/injected/promise_manager.rs @@ -1,8 +1,7 @@ // Copyright (C) Gear Technologies Inc. // SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0 -use anyhow::Result; -use dashmap::{DashMap, mapref::entry::Entry}; +use dashmap::DashMap; use ethexe_common::{ Address, HashOf, db::{ @@ -14,13 +13,21 @@ use ethexe_common::{ }, }; use ethexe_db::Database; -use std::sync::Arc; -use tokio::sync::oneshot; +use std::{sync::Arc, time::Duration}; +use tokio::sync::watch; use tracing::{trace, warn}; -// TODO: #5385. -type PromiseSubscribers = - Arc, oneshot::Sender>>; +/// Bounds how many concurrent watchers a single not-yet-resolved transaction can +/// accumulate, so one client cannot pin unbounded background tasks and map entries +/// on the server by opening many watches for the same transaction. +const MAX_SUBSCRIBERS_PER_TX: usize = 32; + +/// `None` until `dispatch_receipt` publishes the receipt. +type ReceiptSlot = Option>; +pub(crate) type ReceiptWatcher = watch::Receiver; +// One `watch` channel per transaction: subscribers are receiver clones, so a watcher +// cancels by dropping its receiver — no per-subscriber id bookkeeping needed. +type PromiseSubscribers = Arc, watch::Sender>>; type PendingReceiptsCache = moka::sync::Cache, UnfilledPromiseReceipt>; /// The manager for promise subscribers. @@ -33,38 +40,67 @@ pub struct PromiseSubscriptionManager { pending_receipts: PendingReceiptsCache, } -#[derive(Debug, Clone, thiserror::Error)] -pub enum RegisterSubscriberError { - #[error("Subscriber for this transaction already exists, tx_hash={0}")] - AlreadyRegistered(HashOf), +pub enum RegisterSubscriberResult { + Ready(SignedTxReceipt), + Pending(PendingSubscriber), + /// Too many concurrent watchers already registered for this transaction. + TooManyWatchers, } -type TimeoutReceiver = tokio::time::Timeout>; - /// The pending [SignedTxReceipt] subscriber. /// Subscriber will be spawned in separate tokio runtime task and will wait for promise. /// -/// Important: to avoid infinite waiting we wrap [oneshot::Receiver] into [tokio::time::timeout]. +/// Important: to avoid infinite waiting the spawner bounds the wait by `timeout`. +/// +/// Owns a handle to the subscribers map and removes its own entry on `Drop`, +/// so cleanup happens whether the subscriber finishes normally, is released +/// before spawning, or is dropped by future cancellation (e.g. client +/// disconnect) — no call site can forget it. pub struct PendingSubscriber { /// Tx hash waiting promise for. tx_hash: HashOf, - /// Wrapped tx receipt [oneshot::Receiver]. - receiver: TimeoutReceiver, + receiver: ReceiptWatcher, + /// Maximum time to wait for the receipt. + timeout: Duration, + subscribers: PromiseSubscribers, } impl PendingSubscriber { - pub fn new( - db: &Database, + fn new( + manager: &PromiseSubscriptionManager, tx_hash: HashOf, - receiver: oneshot::Receiver, + receiver: ReceiptWatcher, ) -> Self { - let timeout_duration = utils::receipt_waiting_timeout(db); - let receiver = tokio::time::timeout(timeout_duration, receiver); - Self { tx_hash, receiver } + Self { + tx_hash, + receiver, + timeout: utils::receipt_waiting_timeout(&manager.db), + subscribers: manager.subscribers.clone(), + } + } + + pub fn timeout(&self) -> Duration { + self.timeout + } + + pub fn receiver_mut(&mut self) -> &mut ReceiptWatcher { + &mut self.receiver + } + + /// Test-only: clones the receiver so tests can drive the channel directly. + /// The clone keeps the map entry's receiver count above the removal + /// threshold, so `self`'s subsequent `Drop` leaves the entry in place for it. + #[cfg(test)] + pub fn into_parts(self) -> (HashOf, ReceiptWatcher, Duration) { + (self.tx_hash, self.receiver.clone(), self.timeout) } +} - pub fn into_parts(self) -> (HashOf, TimeoutReceiver) { - (self.tx_hash, self.receiver) +impl Drop for PendingSubscriber { + fn drop(&mut self) { + // `self.receiver` hasn't dropped yet, so a count of 1 means it's the last one. + self.subscribers + .remove_if(&self.tx_hash, |_, sender| sender.receiver_count() == 1); } } @@ -77,26 +113,61 @@ impl PromiseSubscriptionManager { } } - // TODO: Issue #5402 pub fn try_register_subscriber( &self, tx_hash: HashOf, - ) -> Result { - match self.subscribers.entry(tx_hash) { - Entry::Occupied(_) => Err(RegisterSubscriberError::AlreadyRegistered(tx_hash)), - Entry::Vacant(entry) => { - let (sender, receiver) = oneshot::channel(); - entry.insert(sender); - Ok(PendingSubscriber::new(&self.db, tx_hash, receiver)) + ) -> RegisterSubscriberResult { + self.try_register_subscriber_inner(tx_hash, || {}) + } + + /// `after_insert` is a test-only seam: it runs between the subscriber-map insert + /// and the race-closing recheck below, so a test can deterministically land a + /// receipt in that exact window (see `race_window_receipt_is_still_served_ready`). + fn try_register_subscriber_inner( + &self, + tx_hash: HashOf, + after_insert: impl FnOnce(), + ) -> RegisterSubscriberResult { + if let Some(receipt) = self.ready_stored_receipt(tx_hash) { + return RegisterSubscriberResult::Ready(receipt); + } + + let receiver = { + let sender = self + .subscribers + .entry(tx_hash) + .or_insert_with(|| watch::channel(None).0); + if sender.receiver_count() >= MAX_SUBSCRIBERS_PER_TX { + return RegisterSubscriberResult::TooManyWatchers; } + sender.subscribe() + }; + + after_insert(); + + // Recheck: `store_and_dispatch_receipt` persists before dispatching, so a receipt + // landing mid-registration is visible here even if its dispatch ran before our insert. + if let Some(receipt) = self.ready_stored_receipt(tx_hash) { + drop(receiver); + // Drops the per-transaction entry, since no live watchers remain. + self.subscribers + .remove_if(&tx_hash, |_, sender| sender.receiver_count() == 0); + return RegisterSubscriberResult::Ready(receipt); } + + RegisterSubscriberResult::Pending(PendingSubscriber::new(self, tx_hash, receiver)) } - pub fn cancel_registration( + /// Stored receipt that can answer a registration, or `None` if there is none yet + /// or the only one stored is a stale `Purged` marker — which must not block a + /// resubmission from getting its own relay. + fn ready_stored_receipt( &self, tx_hash: HashOf, - ) -> Option> { - self.subscribers.remove(&tx_hash).map(|(_, v)| v) + ) -> Option { + self.db + .receipt(tx_hash) + .filter(|receipt| !receipt.data().is_purged()) } // TODO: Issue #5403 @@ -207,10 +278,13 @@ impl PromiseSubscriptionManager { } fn dispatch_receipt(&self, receipt: SignedTxReceipt) { - if let Some((_, sender)) = self.subscribers.remove(&receipt.data().tx_hash()) - && let Err(unsent_receipt) = sender.send(receipt) - { - trace!("failed to send receipt to subscriber, receipt={unsent_receipt:?}"); + let tx_hash = receipt.data().tx_hash(); + let Some((_tx_hash, sender)) = self.subscribers.remove(&tx_hash) else { + return; + }; + + if sender.send(Some(Arc::new(receipt))).is_err() { + trace!(%tx_hash, "no live subscribers left for the receipt"); } } @@ -221,7 +295,10 @@ impl PromiseSubscriptionManager { #[cfg(test)] pub fn subscribers_count(&self) -> usize { - self.subscribers.len() + self.subscribers + .iter() + .map(|entry| entry.value().receiver_count()) + .sum() } } @@ -302,16 +379,22 @@ mod tests { fn register( manager: &PromiseSubscriptionManager, tx_hash: HashOf, - ) -> std::pin::Pin>> { + ) -> ReceiptWatcher { let pending = match manager.try_register_subscriber(tx_hash) { - Ok(pending) => pending, - Err(err) => panic!("first registration must succeed: {err}"), + RegisterSubscriberResult::Pending(pending) => pending, + _ => panic!("empty database must produce a pending registration"), }; - let (_, receiver) = pending.into_parts(); - // Inner oneshot::Receiver is Unpin; the outer Timeout is not, - // hence we discard the timeout wrapper (tests drive their own - // timing via tokio::time::timeout below). - Box::pin(receiver.into_inner()) + let (_tx_hash, receiver, _timeout) = pending.into_parts(); + receiver + } + + async fn next_receipt(receiver: &mut ReceiptWatcher) -> Arc { + receiver + .wait_for(|receipt| receipt.is_some()) + .await + .expect("receipt sender must be alive") + .clone() + .expect("`wait_for` guarantees the receipt is set") } /// Producer signature lands after the local node has already @@ -334,7 +417,7 @@ mod tests { set_current_validators(&db, vec![receipt.address()]); manager.on_tx_receipt(receipt.into()); - let delivered = receiver.as_mut().await.unwrap(); + let delivered = next_receipt(&mut receiver).await; let expected_receipt = Receipt::Promise(promise.clone()); assert_eq!(delivered.data().clone(), expected_receipt); assert_eq!(manager.subscribers_count(), 0); @@ -364,7 +447,7 @@ mod tests { assert_eq!(manager.subscribers_count(), 1); manager.on_computed_promise(promise.clone()); - let delivered = receiver.as_mut().await.unwrap(); + let delivered = next_receipt(&mut receiver).await; let expected_receipt = Receipt::Promise(promise.clone()); assert_eq!(delivered.data(), &Receipt::Promise(promise.clone())); @@ -376,19 +459,6 @@ mod tests { ); } - /// A duplicate registration for the same tx hash is rejected. - #[tokio::test] - async fn duplicate_subscriber_rejected() { - let manager = PromiseSubscriptionManager::new(Database::memory()); - let (promise, _) = make_promise(); - let _first = manager.try_register_subscriber(promise.tx_hash).ok(); - let err = manager - .try_register_subscriber(promise.tx_hash) - .err() - .expect("second registration must fail"); - assert!(matches!(err, RegisterSubscriberError::AlreadyRegistered(_))); - } - /// A compact promise whose signature does not match the body that /// arrives later is parked rather than delivering a malformed /// [`SignedTxReceipt`]. @@ -416,8 +486,11 @@ mod tests { manager.on_tx_receipt(bad_receipt.into()); manager.on_computed_promise(promise.clone()); - let elapsed = - tokio::time::timeout(std::time::Duration::from_millis(50), receiver.as_mut()).await; + let elapsed = tokio::time::timeout( + std::time::Duration::from_millis(50), + receiver.wait_for(|receipt| receipt.is_some()), + ) + .await; assert!(elapsed.is_err(), "no signed promise should be delivered"); assert_eq!(db.promise(tx_hash), Some(promise)); assert_eq!(db.receipt(tx_hash), None); @@ -499,4 +572,127 @@ mod tests { assert_eq!(db.receipt(tx_hash), None); } + + #[tokio::test] + async fn multiple_subscribers_receive_same_receipt() { + let db = Database::memory(); + let manager = PromiseSubscriptionManager::new(db.clone()); + let (promise, private_key) = make_promise(); + let tx_hash = promise.tx_hash; + + let mut first = register(&manager, tx_hash); + let mut second = register(&manager, tx_hash); + + manager.on_computed_promise(promise.clone()); + let receipt = + SignedMessage::create(private_key, Receipt::Promise(promise.to_compact())).unwrap(); + set_current_validators(&db, vec![receipt.address()]); + manager.on_tx_receipt(receipt.into()); + + let first_receipt = next_receipt(&mut first).await; + let second_receipt = next_receipt(&mut second).await; + + assert_eq!(first_receipt.data(), &Receipt::Promise(promise.clone())); + assert_eq!(second_receipt.data(), &Receipt::Promise(promise)); + assert_eq!(manager.subscribers_count(), 0); + } + + #[test] + fn late_subscriber_gets_stored_receipt_without_registration() { + let db = Database::memory(); + let manager = PromiseSubscriptionManager::new(db.clone()); + let (promise, private_key) = make_promise(); + let tx_hash = promise.tx_hash; + let receipt: SignedTxReceipt = + SignedMessage::create(private_key, Receipt::Promise(promise.clone())) + .unwrap() + .into(); + + db.set_receipt(&receipt); + + match manager.try_register_subscriber(tx_hash) { + RegisterSubscriberResult::Ready(ready) => assert_eq!(ready, receipt), + _ => panic!("stored receipt must be returned immediately"), + } + + assert_eq!(manager.subscribers_count(), 0); + } + + #[tokio::test] + async fn release_one_subscriber_keeps_other_subscriber() { + let manager = PromiseSubscriptionManager::new(Database::memory()); + let (promise, _) = make_promise(); + let tx_hash = promise.tx_hash; + + let first = match manager.try_register_subscriber(tx_hash) { + RegisterSubscriberResult::Pending(subscriber) => subscriber, + _ => panic!("empty database must produce a pending registration"), + }; + let second = match manager.try_register_subscriber(tx_hash) { + RegisterSubscriberResult::Pending(subscriber) => subscriber, + _ => panic!("empty database must produce a pending registration"), + }; + + assert_eq!(manager.subscribers_count(), 2); + drop(first); + assert_eq!(manager.subscribers_count(), 1); + drop(second); + assert_eq!(manager.subscribers_count(), 0); + } + + /// Forces a receipt to land in the exact window the second `db.receipt` check in + /// `try_register_subscriber` exists to close (between the subscriber-map insert and + /// that recheck), via the `after_insert` test seam. + #[test] + fn race_window_receipt_is_still_served_ready() { + let db = Database::memory(); + let manager = PromiseSubscriptionManager::new(db.clone()); + let (promise, private_key) = make_promise(); + let tx_hash = promise.tx_hash; + let receipt: SignedTxReceipt = + SignedMessage::create(private_key, Receipt::Promise(promise.clone())) + .unwrap() + .into(); + + let result = manager.try_register_subscriber_inner(tx_hash, || { + db.set_receipt(&receipt); + }); + + match result { + RegisterSubscriberResult::Ready(ready) => assert_eq!(ready, receipt), + _ => panic!("a receipt landing mid-registration must still be served as Ready"), + } + assert_eq!( + manager.subscribers_count(), + 0, + "the race-window registration must be rolled back" + ); + } + + /// A single transaction cannot accumulate unbounded live watchers, and a released + /// watcher frees its slot. + #[tokio::test] + async fn too_many_watchers_for_one_tx_hash_is_rejected() { + let manager = PromiseSubscriptionManager::new(Database::memory()); + let (promise, _) = make_promise(); + let tx_hash = promise.tx_hash; + + let subscribers: Vec<_> = (0..MAX_SUBSCRIBERS_PER_TX) + .map(|_| match manager.try_register_subscriber(tx_hash) { + RegisterSubscriberResult::Pending(subscriber) => subscriber, + _ => panic!("registrations under the cap must be pending"), + }) + .collect(); + + assert!(matches!( + manager.try_register_subscriber(tx_hash), + RegisterSubscriberResult::TooManyWatchers + )); + + drop(subscribers); + assert!(matches!( + manager.try_register_subscriber(tx_hash), + RegisterSubscriberResult::Pending(_) + )); + } } diff --git a/ethexe/rpc/server/src/apis/injected/relay.rs b/ethexe/rpc/server/src/apis/injected/relay.rs index f5d53760010..89378933245 100644 --- a/ethexe/rpc/server/src/apis/injected/relay.rs +++ b/ethexe/rpc/server/src/apis/injected/relay.rs @@ -5,24 +5,45 @@ //! //! [`TransactionsRelayer::relay`] broadcasts a transaction to every //! validator in the current era and returns the first acceptance. +//! +//! Concurrent calls for the same transaction hash share a single in-flight +//! relay — one service event, one network broadcast, one Accept/Reject +//! outcome observed by every caller. The in-flight entry is removed once the +//! outcome is published, so a later resubmission relays afresh. use crate::{RpcEvent, errors}; -use ethexe_common::injected::{InjectedTransactionAcceptance, SignedInjectedTransaction}; +use dashmap::{DashMap, mapref::entry::Entry}; +use ethexe_common::{ + HashOf, + injected::{InjectedTransaction, InjectedTransactionAcceptance, SignedInjectedTransaction}, +}; use jsonrpsee::core::RpcResult; -use tokio::sync::{mpsc, oneshot}; +use std::sync::Arc; +use tokio::sync::{mpsc, oneshot, watch}; + +/// `None` until the shared relay resolves with the service's answer. +type RelayOutcome = Option>; +type InFlightRelays = Arc, watch::Receiver>>; #[derive(Debug, Clone)] pub struct TransactionsRelayer { rpc_sender: mpsc::UnboundedSender, + in_flight: InFlightRelays, } impl TransactionsRelayer { pub fn new(rpc_sender: mpsc::UnboundedSender) -> Self { - Self { rpc_sender } + Self { + rpc_sender, + in_flight: InFlightRelays::default(), + } } /// Broadcast `transaction` to every validator in the current era, /// returning the first `Accept` observed by the service. + /// + /// Deduplicated per transaction hash: while a relay for this hash is in + /// flight, concurrent callers await its outcome instead of relaying again. pub async fn relay( &self, transaction: SignedInjectedTransaction, @@ -41,30 +62,130 @@ impl TransactionsRelayer { )); } - let (response_sender, response_receiver) = oneshot::channel(); - let event = RpcEvent::InjectedTransaction { - transaction, - response_sender, + let mut outcome_rx = match self.in_flight.entry(tx_hash) { + Entry::Occupied(entry) => entry.get().clone(), + Entry::Vacant(entry) => { + let (outcome_tx, outcome_rx) = watch::channel(None); + entry.insert(outcome_rx.clone()); + + let rpc_sender = self.rpc_sender.clone(); + let in_flight = Arc::clone(&self.in_flight); + // Detached so caller cancellation cannot strand other waiters: + // the task always publishes the outcome or drops the sender. + tokio::spawn(async move { + // Guards removal against a panic in `relay_to_service`, not just normal completion. + let _guard = scopeguard::guard(tx_hash, move |tx_hash| { + in_flight.remove(&tx_hash); + }); + let result = relay_to_service(rpc_sender, transaction, tx_hash).await; + // Publish first, or a concurrent caller could see a vacant entry and relay again. + let _ = outcome_tx.send(Some(result)); + }); + outcome_rx + } }; - if let Err(err) = self.rpc_sender.send(event) { - tracing::error!( - "Failed to send `RpcEvent::InjectedTransaction` event task: {err}. \ - The receiving end in the main service might have been dropped." - ); - return Err(errors::internal()); - } + let outcome = outcome_rx + .wait_for(|outcome| outcome.is_some()) + .await + .map_err(|_relay_task_gone| errors::internal())?; + outcome + .as_ref() + .expect("`wait_for` guarantees the outcome is set") + .clone() + } +} + +/// Sends the transaction to the main service and awaits its acceptance. +async fn relay_to_service( + rpc_sender: mpsc::UnboundedSender, + transaction: SignedInjectedTransaction, + tx_hash: HashOf, +) -> RpcResult { + let (response_sender, response_receiver) = oneshot::channel(); + let event = RpcEvent::InjectedTransaction { + transaction, + response_sender, + }; + + if let Err(err) = rpc_sender.send(event) { + tracing::error!( + "Failed to send `RpcEvent::InjectedTransaction` event task: {err}. \ + The receiving end in the main service might have been dropped." + ); + return Err(errors::internal()); + } - tracing::trace!(%tx_hash, "Relayed transaction, waiting for acceptance"); + tracing::trace!(%tx_hash, "Relayed transaction, waiting for acceptance"); - response_receiver.await.map_err(|recv_err| { - tracing::error!( - ?tx_hash, - ?recv_err, - "transaction acceptance channel dropped" + response_receiver.await.map_err(|recv_err| { + tracing::error!( + ?tx_hash, + ?recv_err, + "transaction acceptance channel dropped" + ); + + errors::internal() + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use ethexe_common::{SignedMessage, ecdsa::PrivateKey, mock::Mock}; + + fn mock_signed_transaction() -> SignedInjectedTransaction { + SignedMessage::create(PrivateKey::random(), InjectedTransaction::mock(())).unwrap() + } + + #[tokio::test] + async fn concurrent_relays_of_same_transaction_share_one_relay() { + let (sender, mut receiver) = mpsc::unbounded_channel(); + let relayer = TransactionsRelayer::new(sender); + let tx = mock_signed_transaction(); + + let responder = tokio::spawn(async move { + let RpcEvent::InjectedTransaction { + response_sender, .. + } = receiver.recv().await.expect("relay event"); + response_sender + .send(InjectedTransactionAcceptance::Accept) + .expect("response receiver remains open"); + + let second = + tokio::time::timeout(std::time::Duration::from_millis(200), receiver.recv()).await; + assert!( + second.is_err(), + "same in-flight transaction must not relay twice" ); + }); + + let (first, second) = tokio::join!(relayer.relay(tx.clone()), relayer.relay(tx)); + assert_eq!(first.unwrap(), InjectedTransactionAcceptance::Accept); + assert_eq!(second.unwrap(), InjectedTransactionAcceptance::Accept); + responder.await.unwrap(); + } + + #[tokio::test] + async fn completed_relay_does_not_cache_outcome() { + let (sender, mut receiver) = mpsc::unbounded_channel(); + let relayer = TransactionsRelayer::new(sender); + let tx = mock_signed_transaction(); + + // Two sequential relays must produce two service events. + let responder = tokio::spawn(async move { + for _ in 0..2 { + let RpcEvent::InjectedTransaction { + response_sender, .. + } = receiver.recv().await.expect("relay event"); + response_sender + .send(InjectedTransactionAcceptance::Accept) + .expect("response receiver remains open"); + } + }); - errors::internal() - }) + relayer.relay(tx.clone()).await.unwrap(); + relayer.relay(tx).await.unwrap(); + responder.await.expect("both relays reached the service"); } } diff --git a/ethexe/rpc/server/src/apis/injected/server.rs b/ethexe/rpc/server/src/apis/injected/server.rs index c34040e1fdc..0f4c54a2ff0 100644 --- a/ethexe/rpc/server/src/apis/injected/server.rs +++ b/ethexe/rpc/server/src/apis/injected/server.rs @@ -4,7 +4,9 @@ use crate::{RpcEvent, errors, metrics::InjectedApiMetrics}; use super::{ - InjectedServer, promise_manager::PromiseSubscriptionManager, relay::TransactionsRelayer, + InjectedServer, + promise_manager::{PromiseSubscriptionManager, RegisterSubscriberResult}, + relay::TransactionsRelayer, spawner, }; use ethexe_common::{ @@ -95,7 +97,6 @@ impl InjectedApi { self.relayer.relay(transaction).await } - // TODO: Issue #5386. async fn send_transaction_and_watch( &self, pending: PendingSubscriptionSink, @@ -104,31 +105,44 @@ impl InjectedApi { let tx_hash = transaction.data().to_hash(); let pending_subscriber = match self.manager.try_register_subscriber(tx_hash) { - Ok(subscriber) => subscriber, - Err(err) => { - return Err(errors::bad_request(err).into()); + RegisterSubscriberResult::Ready(receipt) => { + // Not counted in `injected_tx_active_subscriptions`: never enters the pending state. + let sink = pending.accept().await?; + spawner::send_receipt(&sink, &receipt).await; + return Ok(()); + } + RegisterSubscriberResult::TooManyWatchers => { + return Err(errors::bad_request("too many watchers for this transaction").into()); } + RegisterSubscriberResult::Pending(subscriber) => subscriber, }; - let acceptance = self.relayer.relay(transaction).await.inspect_err(|_err| { - self.manager.cancel_registration(tx_hash); - })?; - let sink = match acceptance { - InjectedTransactionAcceptance::Accept => { - pending.accept().await.inspect_err(|_err| { - self.manager.cancel_registration(tx_hash); - })? + // The relayer dedups in-flight relays per tx hash, so concurrent watchers of one + // transaction share a single relay and observe the same Accept/Reject outcome. + let acceptance = match self.relayer.relay(transaction).await { + Ok(acceptance) => acceptance, + Err(err) => { + drop(pending_subscriber); + return Err(err.into()); } + }; + let sink = match acceptance { + InjectedTransactionAcceptance::Accept => match pending.accept().await { + Ok(sink) => sink, + Err(err) => { + drop(pending_subscriber); + return Err(err.into()); + } + }, InjectedTransactionAcceptance::Reject { reason } => { - self.manager.cancel_registration(tx_hash); + drop(pending_subscriber); return Err(reason.into()); } }; self.metrics.injected_tx_active_subscriptions.increment(1); - let (manager, metrics) = (self.manager.clone(), self.metrics.clone()); - spawner::spawn_pending_subscriber(sink, pending_subscriber, move |tx_hash| { - manager.cancel_registration(tx_hash); + let metrics = self.metrics.clone(); + spawner::spawn_pending_subscriber(sink, pending_subscriber, move || { metrics.injected_tx_active_subscriptions.decrement(1); }); Ok(()) diff --git a/ethexe/rpc/server/src/apis/injected/spawner.rs b/ethexe/rpc/server/src/apis/injected/spawner.rs index 34ebe957f45..6cc4b38316d 100644 --- a/ethexe/rpc/server/src/apis/injected/spawner.rs +++ b/ethexe/rpc/server/src/apis/injected/spawner.rs @@ -2,58 +2,71 @@ // SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0 use super::promise_manager::PendingSubscriber; -use ethexe_common::{HashOf, injected::InjectedTransaction}; +use ethexe_common::injected::SignedTxReceipt; use jsonrpsee::{SubscriptionMessage, SubscriptionSink}; use tracing::{error, trace, warn}; +pub(crate) async fn send_receipt(sink: &SubscriptionSink, receipt: &SignedTxReceipt) { + match SubscriptionMessage::from_json(receipt) { + Ok(message) => { + if let Err(err) = sink.send(message).await { + trace!("failed to send receipt, client disconnected: err={err}"); + } + } + Err(err) => { + error!( + ?receipt, + ?err, + "serialization error: failed to create `SubscriptionMessage` from receipt; this must never happen" + ); + } + } +} + /// Spawns [PendingSubscriber] in tokio runtime. /// -/// On task finishing applies the `on_finish` function that is need to drop some data. +/// `subscriber` itself is kept alive for the whole task instead of being torn +/// apart up front: its `Drop` impl removes the subscribers-map entry once its +/// receiver is actually gone, and that stays correct no matter which branch +/// below runs or whether the task is cancelled mid-wait. +/// +/// `on_finish` is a secondary, best-effort hook (currently used for metrics) +/// and is not relied on for map cleanup. pub fn spawn_pending_subscriber( sink: SubscriptionSink, - subscriber: PendingSubscriber, + mut subscriber: PendingSubscriber, on_finish: F, ) where - F: FnOnce(HashOf) + std::marker::Send + 'static, + F: FnOnce() + std::marker::Send + 'static, { - let (tx_hash, receiver) = subscriber.into_parts(); - let _handle = tokio::spawn(async move { - let _guard = scopeguard::guard(tx_hash, on_finish); + let timeout = subscriber.timeout(); - // Waiting for the first one: promise, timeout_err, client disconnect error. + // Waiting for the first one: receipt, timeout_err, client disconnect error. let receipt = tokio::select! { - result = receiver => match result { - Ok(receipt_result) => match receipt_result { - Ok(receipt) => receipt, - Err(_err) => { - unreachable!("promise sender is owned by the server; it cannot be dropped before this point"); - } - }, - Err(_) => { + result = tokio::time::timeout(timeout, subscriber.receiver_mut().wait_for(|receipt| receipt.is_some())) => match result { + Ok(Ok(receipt)) => Some(receipt.clone().expect("`wait_for` guarantees the receipt is set")), + Ok(Err(_sender_dropped)) => { + warn!("receipt sender dropped before delivery, stop background task"); + None + } + Err(_elapsed) => { warn!("promise wasn't received in time, finish waiting"); - return; + None } }, _ = sink.closed() => { trace!("subscription closed by user, stop background task"); - return; + None } }; - match SubscriptionMessage::from_json(&receipt) { - Ok(message) => { - if let Err(err) = sink.send(message).await { - trace!("failed to send promise, client disconnected: err={err}"); - } - } - Err(err) => { - error!( - ?receipt, - ?err, - "serialization error: failed to create `SubscriptionMessage` from receipt; this must never happen" - ); - } + // Free the subscribers-map entry now, before the final send, not at task end. + drop(subscriber); + + if let Some(receipt) = receipt { + send_receipt(&sink, receipt.as_ref()).await; } + on_finish(); }); } diff --git a/ethexe/rpc/server/src/tests.rs b/ethexe/rpc/server/src/tests.rs index 14168f6b597..c4fce83830d 100644 --- a/ethexe/rpc/server/src/tests.rs +++ b/ethexe/rpc/server/src/tests.rs @@ -272,6 +272,232 @@ async fn test_cleanup_promise_subscribers() { } } +#[tokio::test] +#[ntest::timeout(60_000)] +async fn test_same_transaction_multiple_and_late_watchers() { + let listen_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 8014); + let MockService { + mut rpc, + handle, + validator_key, + } = MockService::new(listen_addr).await; + + // Inject promises/receipts through a clone of the API; the server holds the + // same Arc-shared manager, so dispatch reaches the registered subscribers. + // rpc.injected_api is accessible here because tests is a submodule of lib.rs. + let injected_api = rpc.injected_api.clone(); + + // Manual acceptance pump: answers `Accept` only, never produces promises. + let pump = tokio::spawn(async move { + while let Some(RpcEvent::InjectedTransaction { + response_sender, .. + }) = rpc.next().await + { + let _ = response_sender.send(InjectedTransactionAcceptance::Accept); + } + }); + + let first_client = WsClientBuilder::new() + .build(format!("ws://{listen_addr}")) + .await + .expect("first WS client will be created"); + let second_client = WsClientBuilder::new() + .build(format!("ws://{listen_addr}")) + .await + .expect("second WS client will be created"); + + let tx = mock_signed_transaction(); + let tx_hash = tx.data().to_hash(); + + // Two concurrent watchers for the SAME transaction share one in-flight relay. + let (first, second) = tokio::join!( + first_client.send_transaction_and_watch(tx.clone()), + second_client.send_transaction_and_watch(tx.clone()), + ); + let mut first = first.expect("first subscription will be created"); + let mut second = second.expect("second subscription will be created"); + + // Deterministic proof of two *active* pending subscribers before any receipt. + assert_eq!(injected_api.subscribers_count(), 2); + + // Exactly one promise + one receipt, fanned out to both subscribers. + let promise = Promise::mock(tx_hash); + let receipt = + SignedMessage::create(validator_key, Receipt::Promise(promise.to_compact())).unwrap(); + injected_api.on_computed_promise(promise.clone()); + injected_api.on_tx_receipt(receipt.into()); + + let first_receipt = first.next().await.expect("first item").expect("decodes"); + let second_receipt = second.next().await.expect("second item").expect("decodes"); + assert_eq!(first_receipt.data(), &Receipt::Promise(promise.clone())); + assert_eq!(second_receipt.data(), &Receipt::Promise(promise.clone())); + + wait_for_closed_subscriptions(injected_api.clone()).await; + + // Kill the acceptance pump BEFORE the late watcher so the "Ready path does not + // re-relay" guarantee is actually exercised: with no acceptor, an accidental + // relay errors/hangs instead of being silently accepted. Aborting drops the + // pump's captured `rpc` (and thus the relay `mpsc::Receiver`); awaiting the + // join handle ensures that drop has completed before the call below, so a + // regression that relays hits a closed channel rather than racing the drop. + pump.abort(); + let _ = pump.await; + + // Late watcher for the same tx: the receipt is now stored, so it must be served + // from the cached `Ready` path immediately, without relaying. + let mut late = tokio::time::timeout( + std::time::Duration::from_secs(5), + first_client.send_transaction_and_watch(tx.clone()), + ) + .await + .expect("late subscription must not block on a relay") + .expect("late subscription will be created"); + let late_receipt = tokio::time::timeout(std::time::Duration::from_secs(5), late.next()) + .await + .expect("cached receipt must arrive without a relay") + .expect("late item") + .expect("decodes"); + assert_eq!(late_receipt.data(), &Receipt::Promise(promise)); + + handle.stop().expect("RPC server must stop"); +} + +#[tokio::test] +#[ntest::timeout(60_000)] +async fn test_relay_failure_cleans_pending_subscriber() { + let listen_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 8015); + let MockService { rpc, handle, .. } = MockService::new(listen_addr).await; + let injected_api = rpc.injected_api.clone(); + // Dropping `rpc` closes the relay receiver; the ERROR log from relay.rs is expected here. + drop(rpc); + + let client = WsClientBuilder::new() + .build(format!("ws://{listen_addr}")) + .await + .expect("WS client will be created"); + + let result = client + .send_transaction_and_watch(mock_signed_transaction()) + .await; + + assert!( + result.is_err(), + "relay must fail after its receiver is dropped" + ); + assert_eq!(injected_api.subscribers_count(), 0); + handle.stop().expect("RPC server must stop"); +} + +#[tokio::test] +#[ntest::timeout(60_000)] +async fn test_second_watcher_of_same_transaction_does_not_relay_again() { + let listen_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 8016); + let MockService { + mut rpc, + handle, + validator_key, + } = MockService::new(listen_addr).await; + let injected_api = rpc.injected_api.clone(); + + let pump = tokio::spawn(async move { + let RpcEvent::InjectedTransaction { + response_sender, .. + } = rpc.next().await.expect("relay event"); + response_sender + .send(InjectedTransactionAcceptance::Accept) + .expect("response receiver remains open"); + + // The second watcher of the identical tx_hash must NOT relay again: only the + // leader relays, and the follower awaits the leader's published outcome instead. + let second = tokio::time::timeout(std::time::Duration::from_millis(300), rpc.next()).await; + assert!( + second.is_err(), + "second watcher of the same tx_hash must not trigger its own relay" + ); + }); + + let first_client = WsClientBuilder::new() + .build(format!("ws://{listen_addr}")) + .await + .expect("first WS client will be created"); + let second_client = WsClientBuilder::new() + .build(format!("ws://{listen_addr}")) + .await + .expect("second WS client will be created"); + let tx = mock_signed_transaction(); + let tx_hash = tx.data().to_hash(); + + let (first, second) = tokio::join!( + first_client.send_transaction_and_watch(tx.clone()), + second_client.send_transaction_and_watch(tx), + ); + let mut first = first.expect("leader watcher must be accepted"); + let mut second = second.expect("follower watcher must observe the same acceptance"); + pump.await + .expect("pump must observe exactly one relay event"); + assert_eq!(injected_api.subscribers_count(), 2); + + let promise = Promise::mock(tx_hash); + let receipt = + SignedMessage::create(validator_key, Receipt::Promise(promise.to_compact())).unwrap(); + injected_api.on_computed_promise(promise.clone()); + injected_api.on_tx_receipt(receipt.into()); + + let first_receipt = first.next().await.expect("first item").expect("decodes"); + let second_receipt = second.next().await.expect("second item").expect("decodes"); + assert_eq!(first_receipt.data(), &Receipt::Promise(promise.clone())); + assert_eq!(second_receipt.data(), &Receipt::Promise(promise)); + + wait_for_closed_subscriptions(injected_api).await; + handle.stop().expect("RPC server must stop"); +} + +#[tokio::test] +#[ntest::timeout(60_000)] +async fn test_relay_rejection_rejects_all_watchers_of_same_transaction() { + let listen_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 8018); + let MockService { + mut rpc, handle, .. + } = MockService::new(listen_addr).await; + let injected_api = rpc.injected_api.clone(); + + let pump = tokio::spawn(async move { + let RpcEvent::InjectedTransaction { + response_sender, .. + } = rpc.next().await.expect("relay event"); + response_sender + .send(InjectedTransactionAcceptance::Reject { + reason: "rejected by test".into(), + }) + .expect("response receiver remains open"); + }); + + let first_client = WsClientBuilder::new() + .build(format!("ws://{listen_addr}")) + .await + .expect("first WS client will be created"); + let second_client = WsClientBuilder::new() + .build(format!("ws://{listen_addr}")) + .await + .expect("second WS client will be created"); + let tx = mock_signed_transaction(); + + let (first, second) = tokio::join!( + first_client.send_transaction_and_watch(tx.clone()), + second_client.send_transaction_and_watch(tx), + ); + pump.await + .expect("pump must observe exactly one relay event"); + + assert!(first.is_err(), "leader watcher must see the rejection"); + assert!( + second.is_err(), + "follower watcher must see the same rejection instead of relaying itself" + ); + assert_eq!(injected_api.subscribers_count(), 0); + handle.stop().expect("RPC server must stop"); +} + // Setup worker-threads=4 to simulate concurrent clients. #[tokio::test] #[ntest::timeout(120_000)]