PubSub redesign from broadcast to multicast (#373)

This PR reworks the way events are delivered to websocket subscription
handlers.
The main changes:
1. Every subscriber registers itself in the global subscriptions
database, along with event that it's interested in
2. Geyser subsystem, upon receiving a particular event checks whether
there're active subscribers for the event, and sends it only to those,
no filtering is required further.
3. Rust-Protobuf-Rust conversions are disabled to avoid unnecessary
overhead
4. gRPC plugin is completely disabled to avoid unnecessary overhead

<!-- greptile_comment -->

## Greptile Summary

This PR implements a significant architectural change to the PubSub
system, transitioning from a broadcast model to a multicast model for
more efficient event delivery. The changes focus on targeted message
delivery through a global subscriptions database.

- Introduces `SubscriptionsDb` in
`magicblock-geyser-plugin/src/types.rs` for centralized subscription
management using concurrent hashmaps
- Removes gRPC and protobuf conversion overhead by removing
`conversions.rs` and simplifying message handling
- Adds new `notification_builder.rs` with trait-based notification
system for different event types
- Reduces default channel capacity from 250,000 to 1,024 in `config.rs`,
which may impact high-throughput scenarios
- Adds comprehensive test suite in `test-integration/test-pubsub/`
validating subscription functionality for accounts, logs, signatures,
slots and programs



<!-- /greptile_comment -->

---------

Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>

Author: bmuddha

Cargo.lock

@@ -265,8 +265,8 @@ impl AccountsDb {
 
         let used_storage = self.storage.utilized_mmap();
         if let Err(err) = self.snapshot_engine.snapshot(slot, used_storage) {
-            error!(
-                "error taking snapshot at {}, slot {slot}: {err}",
+            warn!(
+                "failed to take snapshot at {}, slot {slot}: {err}",
                 self.snapshot_engine.database_path().display()
             );
         }

magicblock-accounts-db/src/lib.rs

@@ -134,9 +134,9 @@ pub struct MagicValidator {
     sample_performance_service: Option<SamplePerformanceService>,
     commit_accounts_ticker: Option<tokio::task::JoinHandle<()>>,
     remote_account_fetcher_worker: Option<RemoteAccountFetcherWorker>,
-    remote_account_fetcher_handle: Option<thread::JoinHandle<()>>,
+    remote_account_fetcher_handle: Option<tokio::task::JoinHandle<()>>,
     remote_account_updates_worker: Option<RemoteAccountUpdatesWorker>,
-    remote_account_updates_handle: Option<thread::JoinHandle<()>>,
+    remote_account_updates_handle: Option<tokio::task::JoinHandle<()>>,
     remote_account_cloner_worker: Option<
         RemoteAccountClonerWorker<
             BankAccountProvider,
@@ -145,7 +145,7 @@ pub struct MagicValidator {
             AccountDumperBank,
         >,
     >,
-    remote_account_cloner_handle: Option<thread::JoinHandle<()>>,
+    remote_account_cloner_handle: Option<tokio::task::JoinHandle<()>>,
     accounts_manager: Arc<AccountsManager>,
     transaction_listener: GeyserTransactionNotifyListener,
     rpc_service: JsonRpcService,
@@ -692,15 +692,10 @@ impl MagicValidator {
         {
             let cancellation_token = self.token.clone();
             self.remote_account_fetcher_handle =
-                Some(thread::spawn(move || {
-                    create_worker_runtime("remote_account_fetcher_worker")
-                        .block_on(async move {
-                            remote_account_fetcher_worker
-                                .start_fetch_request_processing(
-                                    cancellation_token,
-                                )
-                                .await;
-                        });
+                Some(tokio::spawn(async move {
+                    remote_account_fetcher_worker
+                        .start_fetch_request_processing(cancellation_token)
+                        .await;
                 }));
         }
     }
@@ -711,15 +706,10 @@ impl MagicValidator {
         {
             let cancellation_token = self.token.clone();
             self.remote_account_updates_handle =
-                Some(thread::spawn(move || {
-                    create_worker_runtime("remote_account_updates_worker")
-                        .block_on(async move {
-                            remote_account_updates_worker
-                                .start_monitoring_request_processing(
-                                    cancellation_token,
-                                )
-                                .await
-                        });
+                Some(tokio::spawn(async move {
+                    remote_account_updates_worker
+                        .start_monitoring_request_processing(cancellation_token)
+                        .await
                 }));
         }
     }
@@ -735,15 +725,10 @@ impl MagicValidator {
 
             let cancellation_token = self.token.clone();
             self.remote_account_cloner_handle =
-                Some(thread::spawn(move || {
-                    create_worker_runtime("remote_account_cloner_worker")
-                        .block_on(async move {
-                            remote_account_cloner_worker
-                                .start_clone_request_processing(
-                                    cancellation_token,
-                                )
-                                .await
-                        });
+                Some(tokio::spawn(async move {
+                    remote_account_cloner_worker
+                        .start_clone_request_processing(cancellation_token)
+                        .await
                 }));
         }
         Ok(())
@@ -801,11 +786,3 @@ fn programs_to_load(programs: &[ProgramConfig]) -> Vec<(Pubkey, String)> {
         .map(|program| (program.id, program.path.clone()))
         .collect()
 }
-
-fn create_worker_runtime(thread_name: &str) -> tokio::runtime::Runtime {
-    tokio::runtime::Builder::new_current_thread()
-        .enable_all()
-        .thread_name(thread_name)
-        .build()
-        .unwrap()
-}

magicblock-api/src/magic_validator.rs

@@ -14,10 +14,12 @@ bs58 = { workspace = true }
 expiring-hashmap = { workspace = true }
 geyser-grpc-proto = { workspace = true }
 hostname = { workspace = true }
+flume = "0.11"
 log = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 magicblock-transaction-status = { workspace = true }
+scc = "2.3" 
 solana-geyser-plugin-interface = { workspace = true }
 solana-sdk = { workspace = true }
 spl-token-2022 = { workspace = true, features = ["no-entrypoint"] }
@@ -27,6 +29,7 @@ tokio-util = { workspace = true }
 tonic = { workspace = true, features = ["gzip", "tls", "tls-roots"] }
 tonic-health = { workspace = true }
 
+
 [build-dependencies]
 anyhow = { workspace = true }
 cargo-lock = { workspace = true }

magicblock-geyser-plugin/Cargo.toml

@@ -70,7 +70,7 @@ pub struct ConfigGrpc {
 }
 
 const MAX_DECODING_MESSAGE_SIZE_DEFAULT: usize = 4 * 1024 * 1024;
-const CHANNEL_CAPACITY_DEFAULT: usize = 250_000;
+const CHANNEL_CAPACITY_DEFAULT: usize = 1024;
 const UNARY_CONCURRENCY_LIMIT_DEFAULT: usize = Semaphore::MAX_PERMITS;
 
 impl Default for ConfigGrpc {