feat(event-streaming): API-driven subscription management by mariocynicys · Pull Request #2172 · GLEECBTC/komodo-defi-framework

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.

laruh · 2024-11-26T03:25:03Z

@mariocynicys could you fix conflicts please

as these won't be addressed in this PR

shamardy

Huge Work! All comments are for discussion and non blockers.

A few items to consider:

Similar to how we test RPCs we need to add some integration tests for each streamer type and streaming in general, we may need to implement a SSE reader / EventSource for these tests
Please resolve conflicts
test_zcoin_tx_streaming is failing in wasm
Please open docs issue
Once this PR is merged, we need to check open PRs for streaming coverage like the WalletConnect PR as an example

mm2src/mm2_event_stream/src/streamer.rs

shamardy · 2025-02-04T19:31:00Z

mm2src/mm2_main/src/rpc/streaming_activations/mod.rs

+    // TODO: If the the streamer was already running, it is probably running with different configuration.
+    // We might want to inform the client that the configuration they asked for wasn't applied and return
+    // the active configuration instead?
+    // pub config: Json,


It would still be strange for the first client to subscribe to set the config. I see that you postponed this as multi client support completion is not a high priority currently, we can think about better solutions for this in the future.

mm2src/mm2_main/src/rpc/streaming_activations/balance.rs

shamardy · 2025-02-04T19:37:36Z

mm2src/mm2_main/src/rpc/streaming_activations/balance.rs

+        MmCoinEnum::EthCoin(_) => (),
+        MmCoinEnum::ZCoin(_)
+        | MmCoinEnum::UtxoCoin(_)
+        | MmCoinEnum::Bch(_)
+        | MmCoinEnum::QtumCoin(_)
+        | MmCoinEnum::Tendermint(_) => {
+            if req.config.is_some() {
+                Err(BalanceStreamingRequestError::EnableError(
+                    "Invalid config provided. No config needed".to_string(),
+                ))?
+            }
+        },
+        _ => Err(BalanceStreamingRequestError::CoinNotSupported)?,


I guess we will need this and other streamers for SIA in the future for GUIs to fully support balance streaming c.c. @Alrighttt

added to the streaming tracking list

mm2src/mm2_main/src/rpc/streaming_activations/balance.rs

shamardy · 2025-02-04T21:24:50Z

mm2src/mm2_main/src/rpc/streaming_activations/tx_history.rs

+            let streamer = ZCoinTxHistoryEventStreamer::new(coin.clone());
+            ctx.event_stream_manager.add(client_id, streamer, coin.spawner()).await
+        },
+        _ => Err(TxHistoryStreamingRequestError::CoinNotSupported)?,


SIA will need to be added here as well c.c. @Alrighttt

added to the streaming tracking list

.gitignore

mm2src/coins/z_coin.rs

mm2src/mm2_main/src/rpc/dispatcher/dispatcher.rs

* dev: feat(event-streaming): API-driven subscription management (#2172)

Refactor event streaming to support dynamic client subscriptions over RPC using a unified StreamingManager. The StreamingManager now orchestrates background streamers by initializing a streamer when a client activates it via the RPC API and shutting it down when no longer needed. Legacy fee estimator endpoints have been replaced with streaming RPC methods, and new task manager RPCs for BCH and Tendermint have been added. Additionally, event stream configuration has been migrated from static JSON settings to runtime API initialization for improved flexibility.

* dev: fix(derive_key_from_path): check length of current_key_material (#2356) chore(release): bump mm2 version to 2.4.0-beta (#2346) fix(tests): add additional testnet sepolia nodes to test code (#2358) fix(swaps): maintain legacy compatibility for negotiation messages (#2353) refactor(SwapOps): impl defaults for protocol specific swapops fns (#2354) feat(tpu-v2): provide swap protocol versioning (#2324) feat(wallet): add change mnemonic password rpc (#2317) fix(tpu-v2): fix tpu-v2 wait for payment spend and extract secret (#2261) feat(tendermint): unstaking/undelegation (#2330) fix(utxo-withdraw): get hw ctx only when `PrivKeyPolicy` is trezor (#2333) feat(event-streaming): API-driven subscription management (#2172) fix(hash-types): remove panic, enforce fixed-size arrays (#2279) fix(ARRR): store unconfirmed change output (#2276) feat(tendermint): staking/delegation (#2322) chore(deps): `timed-map` migration (#2247) fix(mem-leak): `running_swap` never shrinks (#2301) chore(dep-bump): libp2p (#2326) refactor(build script): rewrite the main build script (#2319)

* dev: (24 commits) fix(eth-tpu): remove state from funding validation (GLEECBTC#2334) improvement(rpc-server): rpc server dynamic port allocation (GLEECBTC#2342) fix(tests): fix or ignore unstable tests (GLEECBTC#2365) fix(fs): make `filter_files_by_extension` return only files (GLEECBTC#2364) fix(derive_key_from_path): check length of current_key_material (GLEECBTC#2356) chore(release): bump mm2 version to 2.4.0-beta (GLEECBTC#2346) fix(tests): add additional testnet sepolia nodes to test code (GLEECBTC#2358) fix(swaps): maintain legacy compatibility for negotiation messages (GLEECBTC#2353) refactor(SwapOps): impl defaults for protocol specific swapops fns (GLEECBTC#2354) feat(tpu-v2): provide swap protocol versioning (GLEECBTC#2324) feat(wallet): add change mnemonic password rpc (GLEECBTC#2317) fix(tpu-v2): fix tpu-v2 wait for payment spend and extract secret (GLEECBTC#2261) feat(tendermint): unstaking/undelegation (GLEECBTC#2330) fix(utxo-withdraw): get hw ctx only when `PrivKeyPolicy` is trezor (GLEECBTC#2333) feat(event-streaming): API-driven subscription management (GLEECBTC#2172) fix(hash-types): remove panic, enforce fixed-size arrays (GLEECBTC#2279) fix(ARRR): store unconfirmed change output (GLEECBTC#2276) feat(tendermint): staking/delegation (GLEECBTC#2322) chore(deps): `timed-map` migration (GLEECBTC#2247) fix(mem-leak): `running_swap` never shrinks (GLEECBTC#2301) ...

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.

shamardy requested a review from dimxy November 25, 2024 10:49

mariocynicys mentioned this pull request Dec 2, 2024

return an error if transaction history futures initialization fails #1986

Open

mariocynicys linked an issue Dec 4, 2024 that may be closed by this pull request

return an error if transaction history futures initialization fails #1986

Open

shamardy added 2.4.0-beta and removed 2.3.0-beta labels Dec 9, 2024

onur-ozkan added status: pending review and removed under review labels Dec 16, 2024

mariocynicys added 2 commits January 25, 2025 09:21

super merge with origin/dev

4162a8e

convert fixmes to todos

23ac8a6

as these won't be addressed in this PR

smk762 mentioned this pull request Feb 3, 2025

Update KDF streaming docs GLEECBTC/komodo-docs-mdx#411

Open

merge with origin/dev

e30a8e6

shamardy requested review from shamardy and removed request for shamardy February 3, 2025 09:54

shamardy previously approved these changes Feb 4, 2025

View reviewed changes

mariocynicys added 2 commits February 5, 2025 10:01

fix a typo

a508394

merge with origin/dev

6b5899b

mariocynicys dismissed shamardy’s stale review via 6b5899b February 5, 2025 09:35

mariocynicys mentioned this pull request Feb 5, 2025

New RPC methods (streaming) GLEECBTC/komodo-docs-mdx#413

Open

shamardy approved these changes Feb 6, 2025

View reviewed changes

shamardy changed the title ~~feat(event-streaming): new streams (orders, swaps, eth_fee_estimation)~~ feat(event-streaming): API-driven subscription management Feb 6, 2025

shamardy merged commit 927f84b into dev Feb 6, 2025

shamardy deleted the event-streaming branch February 6, 2025 10:34

dimxy pushed a commit that referenced this pull request Feb 6, 2025

Merge branch 'dev' into burn-dex-fee-for-non-kmd

703e887

* dev: feat(event-streaming): API-driven subscription management (#2172)

shamardy mentioned this pull request Apr 16, 2025

chore(release): v2.4.0-beta #2345

Merged

smk762 mentioned this pull request Apr 30, 2025

Enhancements to streaming API #2434

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(event-streaming): API-driven subscription management#2172

feat(event-streaming): API-driven subscription management#2172
shamardy merged 139 commits intodevfrom
event-streaming

mariocynicys commented Jul 22, 2024 •

edited

Loading

Uh oh!

laruh commented Nov 26, 2024

Uh oh!

shamardy left a comment

Uh oh!

Uh oh!

shamardy Feb 4, 2025

Uh oh!

Uh oh!

shamardy Feb 4, 2025

Uh oh!

mariocynicys Feb 6, 2025

Uh oh!

Uh oh!

shamardy Feb 4, 2025

Uh oh!

mariocynicys Feb 6, 2025

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants

Conversation

mariocynicys commented Jul 22, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Breaking Changes:

Uh oh!

laruh commented Nov 26, 2024

Uh oh!

shamardy left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

shamardy Feb 4, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

shamardy Feb 4, 2025

Choose a reason for hiding this comment

Uh oh!

mariocynicys Feb 6, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

shamardy Feb 4, 2025

Choose a reason for hiding this comment

Uh oh!

mariocynicys Feb 6, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants

mariocynicys commented Jul 22, 2024 •

edited

Loading