feat: committor service

Cargo.toml

@@ -15,6 +15,8 @@ members = [
   "magicblock-accounts-db",
   "magicblock-api",
   "magicblock-bank",
+  "magicblock-committor-program",
+  "magicblock-committor-service",
   "magicblock-config",
   "magicblock-core",
   "magicblock-geyser-plugin",
@@ -25,6 +27,8 @@ members = [
   "magicblock-processor",
   "magicblock-pubsub",
   "magicblock-rpc",
+  "magicblock-rpc-client",
+  "magicblock-table-mania",
   "magicblock-tokens",
   "magicblock-transaction-status",
   "magicblock-version",
@@ -55,18 +59,18 @@ assert_matches = "1.5.0"
 async-trait = "0.1.77"
 base64 = "0.21.7"
 bincode = "1.3.3"
+borsh = { version = "1.5.1", features = ["derive", "unstable__schema"] }
+borsh-derive = "1.5.1"
 bs58 = "0.4.0"
 byteorder = "1.5.0"
 cargo-lock = "10.0.0"
-expiring-hashmap = { path = "./utils/expiring-hashmap" }
 conjunto-transwise = { git = "https://github.com/magicblock-labs/conjunto.git", rev = "bf82b45" }
 console-subscriber = "0.2.0"
-isocountry = "0.3.2"
 crossbeam-channel = "0.5.11"
+ed25519-dalek = "1.0.1"
 enum-iterator = "1.5.0"
 env_logger = "0.11.2"
-magic-domain-program = { git = "https://github.com/magicblock-labs/magic-domain-program.git", rev = "eba7644", default-features = false}
-magicblock-delegation-program = { git = "https://github.com/magicblock-labs/delegation-program.git", rev = "4af7f1c" }
+expiring-hashmap = { path = "./utils/expiring-hashmap" }
 fd-lock = "4.0.2"
 fs_extra = "1.3.0"
 futures-util = "0.3.30"
@@ -76,6 +80,7 @@ hostname = "0.4.0"
 http-body-util = "0.1.2"
 hyper = "1.4.1"
 hyper-util = "0.1.9"
+isocountry = "0.3.2"
 itertools = "0.14"
 jsonrpc-core = "18.0.0"
 jsonrpc-core-client = "18.0.0"
@@ -87,21 +92,7 @@ lazy_static = "1.4.0"
 libc = "0.2.153"
 libloading = "0.7.4"
 log = "0.4.20"
-num_cpus = "1.16.0"
-num-derive = "0.4"
-num-format = "0.4.4"
-num-traits = "0.2"
-paste = "1.0"
-prometheus = "0.13.4"
-# Needs to match https://crates.io/crates/solana-storage-bigtable/2.1.13/dependencies
-prost = "0.11.9"
-rand = "0.8.5"
-rayon = "1.10.0"
-rustc_version = "0.4"
-semver = "1.0.22"
-serde = "1.0.217"
-serde_derive = "1.0"
-serde_json = "1.0"
+magic-domain-program = { git = "https://github.com/magicblock-labs/magic-domain-program.git", rev = "eba7644", default-features = false }
 magicblock-account-cloner = { path = "./magicblock-account-cloner" }
 magicblock-account-dumper = { path = "./magicblock-account-dumper" }
 magicblock-account-fetcher = { path = "./magicblock-account-fetcher" }
@@ -111,8 +102,13 @@ magicblock-accounts-api = { path = "./magicblock-accounts-api" }
 magicblock-accounts-db = { path = "./magicblock-accounts-db" }
 magicblock-api = { path = "./magicblock-api" }
 magicblock-bank = { path = "./magicblock-bank" }
+magicblock-committor-program = { path = "./magicblock-committor-program", features = [
+  "no-entrypoint",
+] }
+magicblock-committor-service = { path = "./magicblock-committor-service" }
 magicblock-config = { path = "./magicblock-config" }
 magicblock-core = { path = "./magicblock-core" }
+magicblock-delegation-program = { git = "https://github.com/magicblock-labs/delegation-program.git", rev = "4af7f1c" }
 magicblock-geyser-plugin = { path = "./magicblock-geyser-plugin" }
 magicblock-ledger = { path = "./magicblock-ledger" }
 magicblock-metrics = { path = "./magicblock-metrics" }
@@ -122,10 +118,29 @@ magicblock-processor = { path = "./magicblock-processor" }
 magicblock-program = { path = "./programs/magicblock" }
 magicblock-pubsub = { path = "./magicblock-pubsub" }
 magicblock-rpc = { path = "./magicblock-rpc" }
+magicblock-rpc-client = { path = "./magicblock-rpc-client" }
+magicblock-table-mania = { path = "./magicblock-table-mania" }
 magicblock-tokens = { path = "./magicblock-tokens" }
 magicblock-transaction-status = { path = "./magicblock-transaction-status" }
 magicblock-version = { path = "./magicblock-version" }
+num-derive = "0.4"
+num-format = "0.4.4"
+num-traits = "0.2"
+num_cpus = "1.16.0"
+paste = "1.0"
+prometheus = "0.13.4"
+# Needs to match https://crates.io/crates/solana-storage-bigtable/2.1.13/dependencies
+prost = "0.11.9"
 protobuf-src = "1.1"
+rand = "0.8.5"
+rayon = "1.10.0"
+rustc_version = "0.4"
+rusqlite = { version = "0.34.0", features = ["bundled"] }                                           # bundled sqlite 3.44
+semver = "1.0.22"
+serde = "1.0.217"
+serde_derive = "1.0"
+serde_json = "1.0"
+sha3 = "0.10.8"
 solana-account = { git = "https://github.com/magicblock-labs/solana-account.git", rev = "7bdfefc" }
 solana-accounts-db = { version = "2.2" }
 solana-account-decoder = { version = "2.2" }
@@ -142,33 +157,37 @@ solana-log-collector = { version = "2.2" }
 solana-measure = { version = "2.2" }
 solana-metrics = { version = "2.2" }
 solana-perf = { version = "2.2" }
+solana-program = "2.2"
 solana-program-runtime = { version = "2.2" }
+solana-program-test = "2.2"
 solana-pubkey = { version = "2.2" }
 solana-rayon-threadlimit = { version = "2.2" }
 solana-pubsub-client = { version = "2.2" }
 solana-rpc = "2.2"
 solana-rpc-client = { version = "2.2" }
 solana-rpc-client-api = { version = "2.2" }
 solana-sdk = { version = "2.2" }
-solana-svm = { version = "2.2", features = [ "dev-context-only-utils" ] }
+solana-svm = { version = "2.2", features = ["dev-context-only-utils"] }
 solana-svm-transaction = { version = "2.2" }
 solana-storage-proto = { path = "storage-proto" }
 solana-system-program = { version = "2.2" }
 solana-timings = "2.2"
 solana-transaction-status = { version = "2.2" }
+solana-transaction-status-client-types = "2.2"
 spl-token = "=7.0"
 spl-token-2022 = "=6.0"
+static_assertions = "1.1.0"
 strum = "0.24"
 strum_macros = "0.24"
 tempfile = "3.10.1"
 test-tools = { path = "./test-tools" }
 test-tools-core = { path = "./test-tools-core" }
 thiserror = "1.0.57"
-toml = "0.8.13"
 # Update solana-tokio patch below when updating this version
 tokio = "1.0"
 tokio-stream = "0.1.15"
 tokio-util = "0.7.10"
+toml = "0.8.13"
 # Tonic version 11 conflicts with lower level deps of solana and 0.9.x is the last
 # version that allows prost 0.11.x to be used
 tonic = "0.9.2"
@@ -181,5 +200,5 @@ vergen = "8.3.1"
 # some solana dependencies have solana-storage-proto as dependency
 # we need to patch them with our version, because they use protobuf-src v1.1.0
 # and we use protobuf-src v2.1.1. Otherwise compilation fails
-solana-storage-proto = { path = "./storage-proto" }
 solana-account = { git = "https://github.com/magicblock-labs/solana-account.git", rev = "7bdfefc" }
+solana-storage-proto = { path = "./storage-proto" }

magicblock-account-cloner/Cargo.toml

@@ -16,6 +16,7 @@ magicblock-account-updates = { workspace = true }
 magicblock-account-dumper = { workspace = true }
 magicblock-accounts-api = { workspace = true }
 magicblock-core = { workspace = true }
+magicblock-committor-service = { workspace = true }
 magicblock-metrics = { workspace = true }
 magicblock-mutator = { workspace = true }
 solana-sdk = { workspace = true }
@@ -25,3 +26,6 @@ thiserror = { workspace = true }
 lru = "0.14"
 
 [dev-dependencies]
+magicblock-committor-service = { workspace = true, features = [
+  "dev-context-only-utils",
+] }

magicblock-account-cloner/src/account_cloner.rs

@@ -8,10 +8,11 @@ use futures_util::future::BoxFuture;
 use magicblock_account_dumper::AccountDumperError;
 use magicblock_account_fetcher::AccountFetcherError;
 use magicblock_account_updates::AccountUpdatesError;
+use magicblock_committor_service::error::CommittorServiceResult;
 use magicblock_core::magic_program;
 use solana_sdk::{clock::Slot, pubkey::Pubkey, signature::Signature};
 use thiserror::Error;
-use tokio::sync::oneshot::Sender;
+use tokio::sync::oneshot::{self, Sender};
 
 #[derive(Debug, Clone, Error)]
 pub enum AccountClonerError {
@@ -30,6 +31,9 @@ pub enum AccountClonerError {
     #[error(transparent)]
     AccountDumperError(#[from] AccountDumperError),
 
+    #[error("CommittorSerivceError {0}")]
+    CommittorSerivceError(String),
+
     #[error("ProgramDataDoesNotExist")]
     ProgramDataDoesNotExist,
 
@@ -66,6 +70,20 @@ pub enum AccountClonerUnclonableReason {
     DelegatedAccountsNotClonedWhileHydrating,
 }
 
+pub async fn map_committor_request_result<T>(
+    res: oneshot::Receiver<CommittorServiceResult<T>>,
+) -> AccountClonerResult<T> {
+    res.await
+        .map_err(|err| {
+            // Send request error
+            AccountClonerError::CommittorSerivceError(format!("{:?}", err))
+        })?
+        .map_err(|err| {
+            // Commit error
+            AccountClonerError::CommittorSerivceError(format!("{:?}", err))
+        })
+}
+
 #[derive(Debug, Clone)]
 pub struct AccountClonerPermissions {
     pub allow_cloning_refresh: bool,

magicblock-account-cloner/src/remote_account_cloner_client.rs

@@ -11,6 +11,7 @@ use magicblock_account_dumper::AccountDumper;
 use magicblock_account_fetcher::AccountFetcher;
 use magicblock_account_updates::AccountUpdates;
 use magicblock_accounts_api::InternalAccountProvider;
+use magicblock_committor_service::ChangesetCommittor;
 use solana_sdk::pubkey::Pubkey;
 use tokio::sync::{mpsc::UnboundedSender, oneshot::channel};
 
@@ -25,14 +26,15 @@ pub struct RemoteAccountClonerClient {
 }
 
 impl RemoteAccountClonerClient {
-    pub fn new<IAP, AFE, AUP, ADU>(
-        worker: &RemoteAccountClonerWorker<IAP, AFE, AUP, ADU>,
+    pub fn new<IAP, AFE, AUP, ADU, CC>(
+        worker: &RemoteAccountClonerWorker<IAP, AFE, AUP, ADU, CC>,
     ) -> Self
     where
         IAP: InternalAccountProvider,
         AFE: AccountFetcher,
         AUP: AccountUpdates,
         ADU: AccountDumper,
+        CC: ChangesetCommittor,
     {
         Self {
             clone_request_sender: worker.get_clone_request_sender(),

magicblock-account-cloner/src/remote_account_cloner_worker.rs

@@ -20,6 +20,7 @@ use magicblock_account_dumper::AccountDumper;
 use magicblock_account_fetcher::AccountFetcher;
 use magicblock_account_updates::{AccountUpdates, AccountUpdatesResult};
 use magicblock_accounts_api::InternalAccountProvider;
+use magicblock_committor_service::ChangesetCommittor;
 use magicblock_metrics::metrics;
 use magicblock_mutator::idl::{get_pubkey_anchor_idl, get_pubkey_shank_idl};
 use solana_sdk::{
@@ -36,8 +37,8 @@ use tokio::{
 use tokio_util::sync::CancellationToken;
 
 use crate::{
-    AccountClonerError, AccountClonerListeners, AccountClonerOutput,
-    AccountClonerPermissions, AccountClonerResult,
+    map_committor_request_result, AccountClonerError, AccountClonerListeners,
+    AccountClonerOutput, AccountClonerPermissions, AccountClonerResult,
     AccountClonerUnclonableReason, CloneOutputMap,
 };
 
@@ -96,11 +97,12 @@ impl ValidatorStage {
     }
 }
 
-pub struct RemoteAccountClonerWorker<IAP, AFE, AUP, ADU> {
+pub struct RemoteAccountClonerWorker<IAP, AFE, AUP, ADU, CC> {
     internal_account_provider: IAP,
     account_fetcher: AFE,
     account_updates: AUP,
     account_dumper: ADU,
+    changeset_committor: Arc<CC>,
     allowed_program_ids: Option<HashSet<Pubkey>>,
     blacklisted_accounts: HashSet<Pubkey>,
     payer_init_lamports: Option<u64>,
@@ -118,31 +120,33 @@ pub struct RemoteAccountClonerWorker<IAP, AFE, AUP, ADU> {
 // SAFETY:
 // we never keep references to monitored_accounts around,
 // especially across await points, so this type is Send
-unsafe impl<IAP, AFE, AUP, ADU> Send
-    for RemoteAccountClonerWorker<IAP, AFE, AUP, ADU>
+unsafe impl<IAP, AFE, AUP, ADU, CC> Send
+    for RemoteAccountClonerWorker<IAP, AFE, AUP, ADU, CC>
 {
 }
 // SAFETY:
 // we never produce references to RefCell in monitored_accounts
 // especially not across await points, so this type is Sync
-unsafe impl<IAP, AFE, AUP, ADU> Sync
-    for RemoteAccountClonerWorker<IAP, AFE, AUP, ADU>
+unsafe impl<IAP, AFE, AUP, ADU, CC> Sync
+    for RemoteAccountClonerWorker<IAP, AFE, AUP, ADU, CC>
 {
 }
 
-impl<IAP, AFE, AUP, ADU> RemoteAccountClonerWorker<IAP, AFE, AUP, ADU>
+impl<IAP, AFE, AUP, ADU, CC> RemoteAccountClonerWorker<IAP, AFE, AUP, ADU, CC>
 where
     IAP: InternalAccountProvider,
     AFE: AccountFetcher,
     AUP: AccountUpdates,
     ADU: AccountDumper,
+    CC: ChangesetCommittor,
 {
     #[allow(clippy::too_many_arguments)]
     pub fn new(
         internal_account_provider: IAP,
         account_fetcher: AFE,
         account_updates: AUP,
         account_dumper: ADU,
+        changeset_committor: Arc<CC>,
         allowed_program_ids: Option<HashSet<Pubkey>>,
         blacklisted_accounts: HashSet<Pubkey>,
         payer_init_lamports: Option<u64>,
@@ -162,6 +166,7 @@ where
             account_fetcher,
             account_updates,
             account_dumper,
+            changeset_committor,
             allowed_program_ids,
             blacklisted_accounts,
             payer_init_lamports,
@@ -289,7 +294,7 @@ where
         let stream = stream::iter(account_keys);
         // NOTE: depending on the RPC provider we may get rate limited if we request
         // account states at a too high rate.
-        // I confirmed the the following concurrency working fine:
+        // I confirmed the following concurrency working fine:
         //   Solana Mainnet: 10
         //   Helius: 20
         // If we go higher than this we hit 429s which causes the fetcher to have to
@@ -708,15 +713,27 @@ where
                     });
                 }
 
-                self.do_clone_delegated_account(
+                let sig = self.do_clone_delegated_account(
                     pubkey,
                     // TODO(GabrielePicco): Avoid cloning
                     &Account {
                         lamports: delegation_record.lamports,
                         ..account.clone()
                     },
                     delegation_record,
-                )?
+                )?;
+
+                // Allow the committer service to reserve pubkeys in lookup tables
+                // that could be needed when we commit this account
+                map_committor_request_result(
+                    self.changeset_committor.reserve_pubkeys_for_committee(
+                        *pubkey,
+                        delegation_record.owner,
+                    ),
+                )
+                .await?;
+
+                sig
             }
         };
         // Return the result