feat: run select commands while a node is running [skip-line-limit]

.github/workflows/ci.yml

@@ -473,7 +473,7 @@ jobs:
     strategy:
       matrix:
         # TODO removed base test for now
-        test-suite: [persist]
+        test-suite: [base, persist]
       fail-fast: false
     steps:
       - name: 'Check out the repo'

Cargo.toml

@@ -6,6 +6,7 @@ members = [
   "crates/cli",
   "crates/compute-provider",
   "crates/config",
+  "crates/console", 
   "crates/crypto",
   "crates/data",
   "crates/enclaveup",
@@ -23,11 +24,13 @@ members = [
   "crates/logger",
   "crates/multithread",
   "crates/net", 
-  "crates/zk-prover",
+  "crates/parity-matrix",
+  "crates/polynomial",
   "crates/program-server", 
   "crates/request",
   "crates/safe",
   "crates/sdk",
+  "crates/socket-server",
   "crates/sortition",
   "crates/support-scripts",
   "crates/sync",
@@ -36,9 +39,8 @@ members = [
   "crates/trbfv",
   "crates/utils",
   "crates/wasm",
-  "crates/parity-matrix",
-  "crates/polynomial",
   "crates/zk-helpers",
+  "crates/zk-prover",
 ]
 exclude = [
   "examples/CRISP",
@@ -75,6 +77,7 @@ repository = "https://github.com/gnosisguild/enclave"
 e3-aggregator = { version = "0.1.15", path = "./crates/aggregator" }
 e3-bfv-client = { version = "0.1.15", path = "./crates/bfv-client" }
 e3-config = { version = "0.1.15", path = "./crates/config" }
+e3-console = { version = "0.1.15", path = "./crates/console" }
 e3-ciphernode-builder = { version = "0.1.15", path = "./crates/ciphernode-builder" }
 e3-crypto = { version = "0.1.15", path = "./crates/crypto" }
 e3-data = { version = "0.1.15", path = "./crates/data" }
@@ -96,6 +99,7 @@ e3-logger = { version = "0.1.15", path = "./crates/logger" }
 e3-net = { version = "0.1.15", path = "./crates/net" }
 e3-compute-provider = { version = "0.1.15", path = "./crates/compute-provider" }
 e3-sortition = { version = "0.1.15", path = "./crates/sortition" }
+e3-socket-server = { version = "0.1.15", path = "./crates/socket-server" }
 e3-program-server = { version = "0.1.15", path = "./crates/program-server" }
 e3-polynomial = { version = "0.1.15", path = "./crates/polynomial" }
 e3-support-scripts = { version = "0.1.15", path = "./crates/support-scripts" }

crates/Dockerfile

@@ -48,6 +48,7 @@ COPY crates/cli/Cargo.toml ./cli/Cargo.toml
 COPY crates/ciphernode-builder/Cargo.toml ./ciphernode-builder/Cargo.toml
 COPY crates/compute-provider/Cargo.toml ./compute-provider/Cargo.toml
 COPY crates/config/Cargo.toml ./config/Cargo.toml
+COPY crates/console/Cargo.toml ./console/Cargo.toml
 COPY crates/crypto/Cargo.toml ./crypto/Cargo.toml
 COPY crates/data/Cargo.toml ./data/Cargo.toml
 COPY crates/enclaveup/Cargo.toml ./enclaveup/Cargo.toml
@@ -74,6 +75,7 @@ COPY crates/request/Cargo.toml ./request/Cargo.toml
 COPY crates/safe/Cargo.toml ./safe/Cargo.toml
 COPY crates/sdk/Cargo.toml ./sdk/Cargo.toml
 COPY crates/sortition/Cargo.toml ./sortition/Cargo.toml
+COPY crates/socket-server/Cargo.toml ./socket-server/Cargo.toml
 COPY crates/support-scripts/Cargo.toml ./support-scripts/Cargo.toml
 COPY crates/sync/Cargo.toml ./sync/Cargo.toml
 COPY crates/test-helpers/Cargo.toml ./test-helpers/Cargo.toml

crates/ciphernode-builder/src/ciphernode_builder.rs

@@ -81,6 +81,8 @@ pub struct CiphernodeBuilder {
     zk_backend: Option<ZkBackend>,
     net_config: Option<NetConfig>,
     ignore_address_check: bool,
+    global_shared_store: bool,
+    global_shared_eventstore: bool,
 }
 
 // Simple Net Configuration
@@ -148,6 +150,8 @@ impl CiphernodeBuilder {
             net_config: None,
             zk_backend: None,
             ignore_address_check: false,
+            global_shared_store: false,
+            global_shared_eventstore: false,
         }
     }
 
@@ -311,6 +315,18 @@ impl CiphernodeBuilder {
         self
     }
 
+    /// Share the store this ciphernode uses with socket server commands
+    pub fn with_shared_store(mut self) -> Self {
+        self.global_shared_store = true;
+        self
+    }
+
+    /// Share the eventstore this ciphernode uses with socket server commands
+    pub fn with_shared_eventstore(mut self) -> Self {
+        self.global_shared_eventstore = true;
+        self
+    }
+
     /// Setup net package components.
     pub fn with_net(mut self, peers: Vec<String>, quic_port: u16) -> Self {
         self.net_config = Some(NetConfig::new(peers, quic_port));
@@ -388,21 +404,22 @@ impl CiphernodeBuilder {
                 EventSystem::persisted(log_path, kv_path)
                     .with_event_bus(local_bus)
                     .with_aggregate_config(aggregate_config.clone())
+                    .with_global_shared_store(self.global_shared_store)
             } else {
                 if let Some(ref store) = self.in_mem_store {
                     EventSystem::in_mem_from_store(store)
                         .with_event_bus(local_bus)
                         .with_aggregate_config(aggregate_config.clone())
+                        .with_global_shared_store(self.global_shared_store)
                 } else {
                     EventSystem::in_mem()
                         .with_event_bus(local_bus)
                         .with_aggregate_config(aggregate_config.clone())
+                        .with_global_shared_store(self.global_shared_store)
                 }
             };
-
         let store = event_system.store()?;
-        let eventstore_ts = event_system.eventstore_getter_ts()?;
-        let eventstore_seq = event_system.eventstore_getter_seq()?;
+        let eventstore = event_system.eventstore_reader()?;
         let cipher = &self.cipher;
         let repositories = Arc::new(store.repositories());
         let mut provider_cache =
@@ -456,6 +473,8 @@ impl CiphernodeBuilder {
                 .as_ref()
                 .ok_or_else(|| anyhow::anyhow!("ZK backend is required for threshold keyshare"))?;
 
+            backend.ensure_installed().await?;
+
             // Ensure signer is available before setting up extensions that need it
             let signer = provider_cache.ensure_signer().await?;
 
@@ -539,15 +558,15 @@ impl CiphernodeBuilder {
             (peer_id, interface, channel_bridge)
         };
 
-        setup_net(topic, bus.clone(), eventstore_ts, interface)?;
+        setup_net(topic, bus.clone(), eventstore.ts(), interface)?;
 
         // Run the sync routine
         sync(
             &bus,
             &evm_config,
             &repositories,
             &aggregate_config,
-            &eventstore_seq,
+            &eventstore.seq(),
         )
         .await?;
 

crates/ciphernode-builder/src/event_system.rs

@@ -5,6 +5,8 @@
 // or FITNESS FOR A PARTICULAR PURPOSE.
 
 use crate::get_enclave_event_bus;
+use crate::global_eventstore_cache::{share_eventstore_reader, EventStoreReader};
+use crate::global_store_cache::share_store;
 use actix::{Actor, Addr, Handler, Recipient};
 use anyhow::{anyhow, Result};
 use e3_data::{
@@ -87,6 +89,11 @@ pub struct EventSystem {
     aggregate_config: OnceCell<AggregateConfig>,
     /// Cached EventStoreAddrs for idempotency
     eventstore_addrs: OnceCell<EventStoreAddrs>,
+    /// Global shared store. This can allow commands to access the database while a node is running.
+    global_shared_store: bool,
+    /// Global shared eventstore. This can allow commands to access the eventstore while a node is
+    /// running.
+    global_shared_eventstore: bool,
 }
 
 impl EventSystem {
@@ -108,6 +115,8 @@ impl EventSystem {
             handle: OnceCell::new(),
             aggregate_config: OnceCell::new(),
             eventstore_addrs: OnceCell::new(),
+            global_shared_store: false,
+            global_shared_eventstore: false,
         }
     }
 
@@ -124,6 +133,8 @@ impl EventSystem {
             handle: OnceCell::new(),
             aggregate_config: OnceCell::new(),
             eventstore_addrs: OnceCell::new(),
+            global_shared_store: false,
+            global_shared_eventstore: false,
         }
     }
 
@@ -142,6 +153,8 @@ impl EventSystem {
             handle: OnceCell::new(),
             aggregate_config: OnceCell::new(),
             eventstore_addrs: OnceCell::new(),
+            global_shared_store: false,
+            global_shared_eventstore: false,
         }
     }
 
@@ -165,6 +178,18 @@ impl EventSystem {
         self
     }
 
+    /// Share the store with other processes
+    pub fn with_global_shared_store(mut self, value: bool) -> Self {
+        self.global_shared_store = value;
+        self
+    }
+
+    /// Share the store with other processes
+    pub fn with_global_shared_eventstore(mut self, value: bool) -> Self {
+        self.global_shared_eventstore = value;
+        self
+    }
+
     /// Get the eventbus address
     pub fn eventbus(&self) -> Addr<EventBus<EnclaveEvent>> {
         self.eventbus.get_or_init(get_enclave_event_bus).clone()
@@ -287,20 +312,24 @@ impl EventSystem {
         }
     }
 
-    pub fn eventstore_getter_seq(&self) -> Result<Recipient<EventStoreQueryBy<SeqAgg>>> {
+    pub fn eventstore_reader(&self) -> Result<EventStoreReader> {
         let eventstores = self.eventstore_addrs()?;
-        match &eventstores {
-            EventStoreAddrs::InMem(_) => Ok(self.in_mem_eventstore_router()?.recipient()),
-            EventStoreAddrs::Persisted(_) => Ok(self.persisted_eventstore_router()?.recipient()),
-        }
-    }
+        let reader = match &eventstores {
+            EventStoreAddrs::InMem(_) => {
+                let router = self.in_mem_eventstore_router()?;
+                EventStoreReader::new(router.clone().recipient(), router.recipient())
+            }
+            EventStoreAddrs::Persisted(_) => {
+                let router = self.persisted_eventstore_router()?;
+                EventStoreReader::new(router.clone().recipient(), router.recipient())
+            }
+        };
 
-    pub fn eventstore_getter_ts(&self) -> Result<Recipient<EventStoreQueryBy<TsAgg>>> {
-        let eventstores = self.eventstore_addrs()?;
-        match &eventstores {
-            EventStoreAddrs::InMem(_) => Ok(self.in_mem_eventstore_router()?.recipient()),
-            EventStoreAddrs::Persisted(_) => Ok(self.persisted_eventstore_router()?.recipient()),
+        if self.global_shared_eventstore {
+            share_eventstore_reader(&reader);
         }
+
+        Ok(reader)
     }
 
     /// Get the BusHandle
@@ -337,6 +366,10 @@ impl EventSystem {
             }
         };
 
+        if self.global_shared_store {
+            share_store(&store)
+        }
+
         Ok(store)
     }
 }

crates/ciphernode-builder/src/global_eventstore_cache.rs

@@ -0,0 +1,50 @@
+// SPDX-License-Identifier: LGPL-3.0-only
+//
+// This file is provided WITHOUT ANY WARRANTY;
+// without even the implied warranty of MERCHANTABILITY
+// or FITNESS FOR A PARTICULAR PURPOSE.
+
+use std::sync::OnceLock;
+
+use actix::Recipient;
+use e3_events::{EventStoreQueryBy, SeqAgg, TsAgg};
+
+#[derive(Clone)]
+pub struct EventStoreReader {
+    query_by_seq: Recipient<EventStoreQueryBy<SeqAgg>>,
+    query_by_ts: Recipient<EventStoreQueryBy<TsAgg>>,
+}
+
+impl EventStoreReader {
+    pub fn new(
+        ts: Recipient<EventStoreQueryBy<TsAgg>>,
+        seq: Recipient<EventStoreQueryBy<SeqAgg>>,
+    ) -> Self {
+        Self {
+            query_by_ts: ts,
+            query_by_seq: seq,
+        }
+    }
+
+    pub fn seq(&self) -> Recipient<EventStoreQueryBy<SeqAgg>> {
+        self.query_by_seq.clone()
+    }
+
+    pub fn ts(&self) -> Recipient<EventStoreQueryBy<TsAgg>> {
+        self.query_by_ts.clone()
+    }
+}
+
+// Hold shared eventstore seq - this is a singleton for production only
+static CACHED_EVENTSTORE_READER: OnceLock<EventStoreReader> = OnceLock::new();
+
+/// Save the eventstore to a cache for use by socket commands. This solves the problem of reusing a
+/// commitlog connection while the node is running in start mode. We can use this during node start.
+/// Only the first call to this is shared.
+pub fn share_eventstore_reader(store: &EventStoreReader) {
+    CACHED_EVENTSTORE_READER.get_or_init(|| store.clone());
+}
+
+pub fn get_shared_eventstore() -> Option<EventStoreReader> {
+    CACHED_EVENTSTORE_READER.get().cloned()
+}