feat(event-streaming): new streams (orders, swaps, eth_fee_estimation) #2172

mariocynicys · 2024-07-22T13:37:21Z

This PR aims to support event streaming in some of MM2's API endpoints.

The work in this PR till now is only a refactor of the event streaming code after multiple iterations of discussing how this should end up looking like.

The flow goes as follows:

A client initially opens an SSE connection with MM2 to receive events on. For wasm, since it can't support SSEs, we use a worker to deliver the events to the client (let's consider it as SSE too).
If a client wants to receive some events (balance change, network topology change, new gas estimations, etc...), they need to subscribe to these events (enable these streamers) from the API.
A streamer is a background task that is responsible for firing some events. Initially MM2 doesn't have any streamers running. When a client enables a streamer (= subscribes to a particular event), they start receiving events from that streamer. Until they unsubscribe to them.
If no client is listening to some active streamer, it will shutdown. A new streamer will start again if a client re-subscribes.
(little irrelevant now) Multiple clients subscribing to the same event on MM2 doesn't mean we will have a new streamer for every client. Only one streamer is booted up for the first subscribed client and new subscribing clients will be sharing the same streamer.

The role of the StreamingManager:

Manage different clients (each client has an id cid when it first connects through SSE).
Add a new (cid, sid) pair (translates to: client with id cid wants to listen to the streamer with id sid). If such a streamer with sid is already running, it is just instructed that a new client is gonna be listening along, otherwise, it will be spawned in the background.
Stop listening (cid, sid) pair (= client with cid no longer wants to receive events from streamer with id sid, if no more clients are listening, the streamer will die).
Manage different streamers.
Manage the connections between the streamers (the background thread) and MM2. Streamers either work in a periodic manner (do some job and possibly fire some events then sleep for a while and repeat) or they need to receive a notification from MM2 (e.g. UTXO balance event streaming rely on electrum subscriptions: the electrum server sends a notification to MM2 when an address balance has changed and MM2 in turn sends this notification to the streamer responsible for firing balance events). To send some data to a streamer, StreamingManager::send(streamer_id, arbitrary_data) is used.

Breaking Changes:

No more event_stream_configuration in MM2 json config. Requested streams are initialized and configured dynamically through the API.
Thus event_stream_configuration.access_control_allow_origin config was moved to access_control_allow_origin (one scope out) b9d1218.
And event_stream_configuration.worker_path was changed to event_stream_worker_path (one scope out) ee32fd1.
No more filters are required in the initial SSE setup with MM2. (e.g. mm2.com/event-stream?filter=NETWORK,BALANCE,DATA_NEEDED:datatype -> mm2.com/event-stream?id=1), but a client id is now required (a random u64 identifier for [& generated by] the client) (e.g. mm2.com/event-stream?id=799384531)
How to filter the wanted events then (client gets no events by default)? The client needs to activate specific event streamers through RPC (e.g. stream::enable::balance).
Contrary to the note in the previous point (client gets no events by default), some special events are now delivered to the client even if they never requested/filtered them (this happens using broadcast_all). example of these events are DATA_NEEDED:datatype, and HEARTBEAT should be similar later as well.

(so for keplr integration, only ?id=<RANDOM_U64> must be added to the SSE init request and everything should work. ?filter= shall be removed)
(the <RANDOM_U64> should be stored by the client since it's required in any event streamer activation/deactivation. If for keplr only though, this isn't necessary for now since DATA_NEEDED:datatype event streamer is a special event that's broadcasted to all clients and doesn't require any sort of activation/subscription before hand)

it was confusing why a utxo standard coin was being used instead of the coin we are building, this makes a bit more sense

…erstanding

There was no point of the option wrapping. Simplifying stuff.

also does some housekeeping, file splitting and renaming

remove the intermediate bridge task and replace it with `filter_map` call. also properly implement the shutdown handle

also wraps the streamermanager mutable data inside Arc<RwLock< so to be used in concurrent code. RwLock was chosen here since we should be sending streaming data way more than editing the streamers

along with some cleanups and simplifications

Also use the `data_rx` for utxo balance events. A cleanup for the scripthash notification channels/logic will follow

this is very messy though (because of the current state of electurm code) and will be reverted, just recording this in the history for now

…nterface" This reverts commit 8972822.

made more sense to have it inside the manager. also now the manager provides the controller as a parameter in the handle method so different streamers don't have to keep a copy of ctx just to broadcast on the controller.

this is still not hooked up though. no data will be received on this rx channel till we do `event_stream_manager.send('BALANCE:ZCoin') (or whatever the coin name is)` somewhere.

mm2src/coins/eth/eth_balance_events.rs

mm2src/mm2_event_stream/src/manager.rs

mm2src/coins/utxo/rpc_clients.rs

compiles but streaming hasn't yet been enabled for the electrum coins

the streaming manager is passed up to ElectrumClientImpl, we could thoeretically pass it down the tree to each ElectrumConnection, but since these connections already have a ref to ElectrumClientImpl in their establish_connecion_loop lifetime, we don't really need to do so.

this commit makes it so that the client deletion logic is handled by the streaming manager when initializing the client. the return type of client initialization has an extra field that will run the client delection logic on Drop. this fixes the problem with clients never getting removed as when the clients disconnect and the body stream is no longer drived further thus will not hit the `remove_client` call.

ignoring the non-wasm test for now since it needs zcash params downloaded which we don't have an auto way to do. would be better if we can download them in the same way we do for wasm (right from kdf).

naming was used loosely, pirate was used as if it was zombie, a previous fix in the url constants led to this breaking.

onur-ozkan · 2024-11-06T08:29:17Z

mm2src/coins/utxo/qtum.rs

@@ -903,7 +903,9 @@ impl MarketCoinOps for QtumCoin {
 impl MmCoin for QtumCoin {
    fn is_asset_chain(&self) -> bool { utxo_common::is_asset_chain(&self.utxo_arc) }

-    fn spawner(&self) -> CoinFutSpawner { CoinFutSpawner::new(&self.as_ref().abortable_system) }
+    fn get_ctx(&self) -> Option<MmArc> { MmArc::from_weak(&self.as_ref().ctx) }


I'm not sure if I mentioned this before (since my last review was a month ago), but why are we implementing this at the trait level? Coins don't create MmArc instances—they're already available at higher levels and some coins just receive them during initialization. We could likely pass MmArc where needed, as we do for certain coins. Adding this function seems unnecessary and inconsistent across different coins.

needed in MmCoin so we can access it in generic impls over MmCoins, e.g.:

komodo-defi-framework/mm2src/coins/utxo/utxo_tx_history_v2.rs

Line 593 in 547fc2b

let streaming_manager = ctx.coin.get_ctx().unwrap().event_stream_manager.clone();

i think i have answered that somewhere in the threads above though, could you please go over it and resolve addressed comments.

p.s. in the e.g. above, we could add it in the UtxoCoinOps instead, but this completely makes no sense IMO. We could also opt for making it get_streaming_manager but i opted for get_ctx since we already have global states other than just the streaming manager and some coins already accesses this global state.

Why it's not in

komodo-defi-framework/mm2src/coins/utxo/utxo_tx_history_v2.rs

Lines 145 to 152 in 547fc2b

struct UtxoTxHistoryStateMachine<Coin: UtxoTxHistoryOps, Storage: TxHistoryStorage> {

coin: Coin,

storage: Storage,

metrics: MetricsArc,

/// Last requested balances of the activated coin's addresses.

/// TODO add a `CoinBalanceState` structure and replace [`HashMap<String, BigDecimal>`] everywhere.

balances: HashMap<String, BigDecimal>,

}

?

that's a good place as well. im down to storing it (streaming manager) there if u are keen on the coins to expose no ctx :(

I'd prefer storing it where it's needed rather than forcing it into an unrelated trait.

into an unrelated trait

gonna have to disagree with that, but guess that's a topic for another day :)

addressed in 65bd662 & e65ace0

pass the streaming manager down to where it should be used instead this does so for utxo balance history, tendermint to come

borngraced

next iteration. I tried to focus on optimizations in this review.

borngraced · 2024-11-15T15:50:05Z

mm2src/mm2_event_stream/src/streamer.rs

+
+    /// Returns a human readable unique identifier for the event streamer.
+    /// No other event streamer should have the same identifier.
+    fn streamer_id(&self) -> String;


streamer_id is being cloned too often, please convert to Arc<str>

borngraced · 2024-11-15T16:24:27Z

mm2src/mm2_event_stream/src/streamer.rs

+            future::ready(
+                if let Ok(input_data) = any_input_data.downcast() {
+                    Some(*input_data)
+                } else {
+                    error!("Couldn't downcast a received message to {}. This message wasn't intended to be sent to this streamer ({streamer_id}).", any::type_name::<<S as EventStreamer>::DataInType>());
+                    None
+                }
+            )


Suggested change

future::ready(

if let Ok(input_data) = any_input_data.downcast() {

Some(*input_data)

} else {

error!("Couldn't downcast a received message to {}. This message wasn't intended to be sent to this streamer ({streamer_id}).", any::type_name::<<S as EventStreamer>::DataInType>());

None

}

)

future::ready(

any_input_data

.downcast()

.map(|input_data| *input_data)

.map_err(|_| { error!(

"Couldn't downcast a received message to {}. This message wasn't intended to be sent to this streamer ({streamer_id}).",

any::type_name::<S::DataInType>()

);

}).ok()

)

borngraced · 2024-11-15T16:30:06Z

mm2src/mm2_event_stream/src/streamer.rs

+    })?;
+
+    // If the handler takes no input data, return `None` for the data sender.
+    if any::TypeId::of::<<S as EventStreamer>::DataInType>() == any::TypeId::of::<NoDataIn>() {


This can be relaxed a lil bit

Suggested change

if any::TypeId::of::<<S as EventStreamer>::DataInType>() == any::TypeId::of::<NoDataIn>() {

if any::TypeId::of::<S::DataInType>() == any::TypeId::of::<NoDataIn>() {

borngraced · 2024-11-15T16:36:56Z

mm2src/mm2_event_stream/src/event.rs

+#[derive(Default)]
+pub struct Event {
+    /// The type of the event (balance, network, swap, etc...).
+    event_type: String,


Can this be converted to Arc<str> ? I see it's mostly constructed via streamer_id which I also suggested to be made into an Arc as it's being cloned too often.

borngraced · 2024-11-15T16:39:13Z

mm2src/mm2_event_stream/src/event.rs

+    pub fn get(&self) -> (String, Json) {
+        let prefix = if self.error { "ERROR:" } else { "" };
+        (format!("{prefix}{}", self.event_type), self.message.clone())
+    }


please return Json as ref and avoid cloning

borngraced · 2024-11-15T16:45:18Z

mm2src/mm2_event_stream/src/manager.rs

+        {
+            let mut this = self.write();
+            match this.clients.get(&client_id) {
+                // We don't know that client. We don't have a connection to it.
+                None => return Err(StreamingManagerError::UnknownClient),
+                // The client is already listening to that streamer.
+                Some(client_info) if client_info.listens_to(&streamer_id) => {
+                    return Err(StreamingManagerError::ClientAlreadyListening);
+                },
+                _ => (),
+            }
+
+            // If a streamer is already up and running, we won't spawn another one.
+            if let Some(streamer_info) = this.streamers.get_mut(&streamer_id) {
+                // Register the client as a listener to the streamer.
+                streamer_info.add_client(client_id);
+                // Register the streamer as listened-to by the client.
+                if let Some(client_info) = this.clients.get_mut(&client_id) {
+                    client_info.add_streamer(streamer_id.clone());
+                }
+                return Ok(streamer_id);
+            }
+        }


did you think about read first” is to avoid unnecessarily taking a write lock when a read lock suffices or is this on purpose ?

borngraced · 2024-11-15T16:56:26Z

mm2src/mm2_event_stream/src/manager.rs

+        let mut this = self.write();
+        if this.clients.contains_key(&client_id) {
+            return Err(StreamingManagerError::ClientExists);
+        }


same as above.

borngraced · 2024-11-15T16:57:54Z

mm2src/mm2_event_stream/src/manager.rs

+        let mut this = self.write();
+        let Some(streamer_info) = this.streamers.get(streamer_id) else { return };
+        if !streamer_info.is_down() {
+            return;
+        }


same as above

borngraced · 2024-11-15T16:59:25Z

mm2src/mm2_event_stream/src/manager.rs

+        // Only `try_send` here. If the channel is full (client is slow), the message
+        // will be dropped and the client won't receive it.
+        // This avoids blocking the broadcast to other receivers.
+        self.channel.try_send(event).ok();


why not self.channel.try_send(event).error_log()

mariocynicys added 30 commits July 10, 2024 11:34

use a different struct to implement the balance streaming for utxo

7642e69

it was confusing why a utxo standard coin was being used instead of the coin we are building, this makes a bit more sense

rename UtxoBalanceStreaming -> UtxoBalanceEventStreamer

8aee75b

add fixme: reminder to send sse in running swap

94fd9b0

refactor(mm2_event_stream::controller): use less types for easier und…

c6ced79

…erstanding

add a note about the broadcasting mechanisim

fc7e232

Take event_stream_configuration out of the Option<

876226b

There was no point of the option wrapping. Simplifying stuff.

don't launch wasm event streamer if there are zero events

58d436c

don't error if in native mode while balance streaming enabled

eaca49b

generalize event streaming config to be any json config

b893483

rework the current events to work with arbitrary input config

737de5b

add always_send parameter to network event config

929279f

balance is a better naming, no need for serde_rename now

4456c74

use a common empty struct for deser-ing empty config

1b9aaa5

add smart filter object to events

9504df4

get rid of error_event_name

d113c6f

also does some housekeeping, file splitting and renaming

rename event behavior to event streamer

209df5f

new event streaming arch

4d15a66

event streamer optimizations and fixmes

009d206

remove the intermediate bridge task and replace it with `filter_map` call. also properly implement the shutdown handle

fix not printing streamer id on downcast error

7a6c224

refuse to add a streamer that is already present

8fa5950

also wraps the streamermanager mutable data inside Arc<RwLock< so to be used in concurrent code. RwLock was chosen here since we should be sending streaming data way more than editing the streamers

adapt network and heartbeat streamers to the new interface

38e22c3

along with some cleanups and simplifications

adapt the rest of the streamers to the new arch

945463f

abandon CoinFutSpawner & MmFutSpawner in favor of WeakSpawner

700e935

fix compilation issues

8af7abc

StreamHandlerInput must be Unpin to call next()

3ade9a2

Also use the `data_rx` for utxo balance events. A cleanup for the scripthash notification channels/logic will follow

PoC: utxo balance streaming using the new streaming manager interface

8972822

this is very messy though (because of the current state of electurm code) and will be reverted, just recording this in the history for now

Revert "PoC: utxo balance streaming using the new streaming manager i…

bf54981

…nterface" This reverts commit 8972822.

hide the Controller inside the StreamingManager

cf50389

made more sense to have it inside the manager. also now the manager provides the controller as a parameter in the handle method so different streamers don't have to keep a copy of ctx just to broadcast on the controller.

fix wasm compilation

1783c2c

use fn handle's data_rx in zcoin balance event streaming

5655db6

this is still not hooked up though. no data will be received on this rx channel till we do `event_stream_manager.send('BALANCE:ZCoin') (or whatever the coin name is)` somewhere.

dimxy reviewed Oct 16, 2024

View reviewed changes

mm2src/coins/eth/eth_balance_events.rs Show resolved Hide resolved

mariocynicys added 2 commits October 19, 2024 11:48

add a todo regarding orderbook streaming in no-login mode

d84bf39

review(dimxy): fix streamer spawner doc comment

52af46f

dimxy mentioned this pull request Oct 21, 2024

Add/gas fee estimator docs KomodoPlatform/komodo-docs-mdx#293

Closed

dimxy reviewed Oct 22, 2024

View reviewed changes

mm2src/mm2_event_stream/src/manager.rs Show resolved Hide resolved

dimxy reviewed Oct 23, 2024

View reviewed changes

mm2src/coins/utxo/rpc_clients.rs Outdated Show resolved Hide resolved

mariocynicys added 3 commits October 30, 2024 19:11

merge with origin/dev

6d6b5a2

compiles but streaming hasn't yet been enabled for the electrum coins

mariocynicys force-pushed the event-streaming branch from 5d72fb9 to c2f0244 Compare October 30, 2024 21:55

tests(streaming-manager): fix client handle dropped early

c29cf81

mariocynicys force-pushed the event-streaming branch 6 times, most recently from dd5ac82 to 39fa8b1 Compare November 1, 2024 18:10

mariocynicys added 2 commits November 1, 2024 20:17

z_coin wasm tx_history streaming test

2a5f66c

ignoring the non-wasm test for now since it needs zcash params downloaded which we don't have an auto way to do. would be better if we can download them in the same way we do for wasm (right from kdf).

fix activate_z_coin_*

159445b

naming was used loosely, pirate was used as if it was zombie, a previous fix in the url constants led to this breaking.

mariocynicys force-pushed the event-streaming branch 2 times, most recently from 0d3ac5f to 159445b Compare November 2, 2024 09:17

mariocynicys added 2 commits November 2, 2024 22:09

add a fixme; check me out

853bbf5

review(onur): cross test event streaming crate

78cece1

smk762 mentioned this pull request Nov 5, 2024

Add EIP1559 Fee management methods and configs KomodoPlatform/komodo-docs-mdx#375

Open

onur-ozkan reviewed Nov 6, 2024

View reviewed changes

mariocynicys added 2 commits November 6, 2024 10:55

review(onur): don't expose ctx from MmCoin

65bd662

pass the streaming manager down to where it should be used instead this does so for utxo balance history, tendermint to come

review(onur): do the same for tendermint history state machine

e65ace0

mariocynicys force-pushed the event-streaming branch from 3a467dd to e65ace0 Compare November 6, 2024 09:55

shamardy added the P2 label Nov 11, 2024

borngraced requested changes Nov 15, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(event-streaming): new streams (orders, swaps, eth_fee_estimation) #2172

feat(event-streaming): new streams (orders, swaps, eth_fee_estimation) #2172

mariocynicys commented Jul 22, 2024 •

edited

Loading

onur-ozkan Nov 6, 2024

mariocynicys Nov 6, 2024 •

edited

Loading

onur-ozkan Nov 6, 2024

mariocynicys Nov 6, 2024

onur-ozkan Nov 6, 2024

mariocynicys Nov 6, 2024

borngraced left a comment

borngraced Nov 15, 2024 •

edited

Loading

borngraced Nov 15, 2024

borngraced Nov 15, 2024

borngraced Nov 15, 2024

borngraced Nov 15, 2024

borngraced Nov 15, 2024

borngraced Nov 15, 2024

borngraced Nov 15, 2024

borngraced Nov 15, 2024

	struct UtxoTxHistoryStateMachine<Coin: UtxoTxHistoryOps, Storage: TxHistoryStorage> {
	coin: Coin,
	storage: Storage,
	metrics: MetricsArc,
	/// Last requested balances of the activated coin's addresses.
	/// TODO add a `CoinBalanceState` structure and replace [`HashMap<String, BigDecimal>`] everywhere.
	balances: HashMap<String, BigDecimal>,
	}

	if any::TypeId::of::<<S as EventStreamer>::DataInType>() == any::TypeId::of::<NoDataIn>() {
	if any::TypeId::of::<S::DataInType>() == any::TypeId::of::<NoDataIn>() {

feat(event-streaming): new streams (orders, swaps, eth_fee_estimation) #2172

Are you sure you want to change the base?

feat(event-streaming): new streams (orders, swaps, eth_fee_estimation) #2172

Conversation

mariocynicys commented Jul 22, 2024 • edited Loading

Breaking Changes:

Choose a reason for hiding this comment

mariocynicys Nov 6, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

borngraced left a comment

Choose a reason for hiding this comment

borngraced Nov 15, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mariocynicys commented Jul 22, 2024 •

edited

Loading

mariocynicys Nov 6, 2024 •

edited

Loading

borngraced Nov 15, 2024 •

edited

Loading