confgi

.dockerignore

@@ -1,2 +1,3 @@
 **target/
 **direnv/
+**vendor/

Dockerfile.operator

@@ -0,0 +1,38 @@
+FROM --platform=$BUILDPLATFORM rust:1.72.0-bookworm as build
+ARG TARGETARCH
+
+WORKDIR /near
+
+COPY rust-toolchain.toml ./rust-toolchain.toml
+COPY Cargo.toml Cargo.lock ./
+COPY nearx/contract near/contract
+
+RUN rustup show 
+RUN apt-get update && apt-get install -y \
+    git \
+    jq \
+    make \
+    bash \
+    openssl \
+    libssl-dev \
+    protobuf-compiler \
+    pkg-config \
+    cbindgen
+
+RUN cargo new --bin bin/dummy
+RUN cargo new --lib crates/dummylib
+RUN cargo new --lib nearx
+RUN cargo build --release --config net.git-fetch-with-cli=true --bin dummy
+RUN rm -rf nearx bin crates
+
+COPY ./ ./
+RUN cargo build --release --config net.git-fetch-with-cli=true --bin nearx-operator
+RUN ldd target/release/nearx-operator 
+RUN cp target/release/nearx-operator /near/near-light-client
+
+FROM debian:bookworm-slim
+RUN apt-get update && apt-get install -y openssl libssl-dev pkg-config ca-certificates && rm -rf /var/lib/apt/lists/*
+COPY --from=build /near/target/release/nearx-operator /usr/local/bin
+COPY --from=build /near/default.toml /var/light-client.toml
+
+ENTRYPOINT ["/usr/local/bin/nearx-operator"]

bin/operator/src/config.rs

@@ -8,6 +8,7 @@ pub struct Config {
     pub rpc: near_light_client_rpc::Config,
     pub protocol: near_light_client_protocol::config::Config,
     pub succinct: crate::succinct::Config,
+    pub engine: crate::engine::Config,
 }
 
 impl Configurable for Config {}

bin/operator/src/engine/mod.rs

@@ -5,7 +5,7 @@ use anyhow::{anyhow, Result};
 use futures::FutureExt;
 use hashbrown::{hash_map::DefaultHashBuilder, HashMap};
 use near_light_client_rpc::{prelude::Itertools, TransactionOrReceiptId};
-use near_light_clientx::VERIFY_AMT;
+use near_light_clientx::config::bps_from_network;
 use priority_queue::PriorityQueue;
 use serde::{Deserialize, Serialize};
 pub use types::RegistryInfo;
@@ -19,22 +19,43 @@ use crate::succinct::{
 
 mod types;
 
-// TODO: decide if we can try to identity hash based on ids, they're already
-// hashed
-// Collision <> receipt & tx?
+// TODO[Optimisation]: decide if we can try to identity hash based on ids,
+// they're already hashed, perhaps a collision would be if receipt_id ++ tx_id
+// are the same, unlikely
 type Queue = PriorityQueue<TransactionOrReceiptIdNewtype, PriorityWeight, DefaultHashBuilder>;
 
+#[derive(Debug, Deserialize, Clone)]
+#[serde(default)]
+pub struct Config {
+    drain_interval: u64,
+    sync_interval: u64,
+    cleanup_interval: u64,
+    persist_interval: u64,
+}
+
+impl Default for Config {
+    fn default() -> Self {
+        Config {
+            drain_interval: 1,
+            sync_interval: 60 * 30,
+            cleanup_interval: 60,
+            persist_interval: 30,
+        }
+    }
+}
+
 pub struct Engine {
     registry: HashMap<usize, RegistryInfo>,
     succinct_client: Arc<succinct::Client>,
-    // TODO: persist me
     proving_queue: Queue,
     batches: HashMap<u32, Option<ProofId>>,
     request_info: HashMap<ProofId, Option<ProofStatus>>,
+    config: Config,
+    verify_amt: usize,
 }
 
 impl Engine {
-    pub fn new(succinct_client: Arc<succinct::Client>) -> Self {
+    pub fn new(config: &super::Config, succinct_client: Arc<succinct::Client>) -> Self {
         log::info!("starting queue manager");
 
         let state = PersistedState::try_from("state.json");
@@ -54,6 +75,8 @@ impl Engine {
                 .map(|s| s.batches.clone())
                 .unwrap_or_default(),
             request_info: state.map(|s| s.request_info).unwrap_or_default(),
+            config: config.engine.clone(),
+            verify_amt: bps_from_network(&config.rpc.network),
         }
     }
 
@@ -70,43 +93,44 @@ impl Engine {
         } else {
             1
         };
-        log::debug!("adding to {:?} with weight: {weight}", tx);
+        log::debug!("enqueuing {:?} with weight: {weight}", tx);
         self.proving_queue.push(tx.into(), weight);
         Ok(())
     }
 
     fn make_batch(&mut self) -> Option<(u32, Vec<TransactionOrReceiptId>)> {
-        if self.proving_queue.len() >= VERIFY_AMT {
-            let id = self.batches.len() as u32;
-            let mut txs = vec![];
-            for _ in 0..VERIFY_AMT {
-                let (req, _) = self.proving_queue.pop().unwrap();
-                txs.push(req.0);
-            }
-            self.batches.insert(id, None);
-            Some((id, txs))
-        } else {
-            None
+        if self.proving_queue.len() < self.verify_amt {
+            return None;
+        }
+        let id = self.batches.len() as u32;
+        let mut txs = vec![];
+        for _ in 0..self.verify_amt {
+            let (req, _) = self.proving_queue.pop()?;
+            txs.push(req.0);
         }
+        self.batches.insert(id, None);
+        Some((id, txs))
     }
 }
 
 impl Actor for Engine {
     type Context = Context<Self>;
 
     fn started(&mut self, ctx: &mut Self::Context) {
-        ctx.run_interval(Duration::from_secs(1), |_, ctx| {
+        ctx.run_interval(Duration::from_secs(self.config.drain_interval), |_, ctx| {
             ctx.address().do_send(Drain)
         });
-        ctx.run_interval(Duration::from_secs(60 * 30), |_, ctx| {
+        ctx.run_interval(Duration::from_secs(self.config.sync_interval), |_, ctx| {
             ctx.address().do_send(Sync)
         });
-        ctx.run_interval(Duration::from_secs(60), |_, ctx| {
-            ctx.address().do_send(Cleanup)
-        });
-        ctx.run_interval(Duration::from_secs(60), |_, ctx| {
-            ctx.address().do_send(Persist)
-        });
+        ctx.run_interval(
+            Duration::from_secs(self.config.cleanup_interval),
+            |_, ctx| ctx.address().do_send(Cleanup),
+        );
+        ctx.run_interval(
+            Duration::from_secs(self.config.persist_interval),
+            |_, ctx| ctx.address().do_send(Persist),
+        );
     }
 }
 
@@ -311,6 +335,8 @@ mod tests {
     use super::*;
     use crate::succinct::tests::mocks;
 
+    const VERIFY_AMT: usize = 64;
+
     async fn manager() -> Engine {
         let client = mocks().await;
         Engine::new(Arc::new(client))

bin/operator/src/engine/types.rs

@@ -65,3 +65,51 @@ pub struct RegistryInfo {
     // Their weight in the shared queue
     pub weight: PriorityWeight,
 }
+
+#[cfg(test)]
+mod tests {
+
+    use std::str::FromStr;
+
+    use near_light_client_protocol::near_account_id::AccountId;
+    use test_utils::CryptoHash;
+
+    use super::*;
+
+    #[test]
+    fn test_transaction_or_receipt_id_eq() {
+        let transaction1 = TransactionOrReceiptIdNewtype(TransactionOrReceiptId::Transaction {
+            transaction_hash: CryptoHash::default(),
+            sender_id: AccountId::from_str("sender1").unwrap(),
+        });
+        let transaction2 = TransactionOrReceiptIdNewtype(TransactionOrReceiptId::Transaction {
+            transaction_hash: CryptoHash::default(),
+            sender_id: AccountId::from_str("sender1").unwrap(),
+        });
+        assert!(transaction1 == transaction2);
+
+        let receipt1 = TransactionOrReceiptIdNewtype(TransactionOrReceiptId::Receipt {
+            receipt_id: CryptoHash::default(),
+            receiver_id: AccountId::from_str("receiver1").unwrap(),
+        });
+        let receipt2 = TransactionOrReceiptIdNewtype(TransactionOrReceiptId::Receipt {
+            receipt_id: CryptoHash::default(),
+            receiver_id: AccountId::from_str("receiver1").unwrap(),
+        });
+        assert!(receipt1 == receipt2);
+
+        let transaction3 = TransactionOrReceiptIdNewtype(TransactionOrReceiptId::Transaction {
+            transaction_hash: CryptoHash::default(),
+            sender_id: AccountId::from_str("sender2").unwrap(),
+        });
+        assert!(transaction1 != transaction3);
+
+        let receipt3 = TransactionOrReceiptIdNewtype(TransactionOrReceiptId::Receipt {
+            receipt_id: CryptoHash::default(),
+            receiver_id: AccountId::from_str("receiver2").unwrap(),
+        });
+        assert!(receipt1 != receipt3);
+
+        assert!(transaction1 != receipt1);
+    }
+}

bin/operator/src/main.rs

@@ -15,7 +15,7 @@ pub async fn main() -> anyhow::Result<()> {
 
     let client = Arc::new(SuccinctClient::new(&config).await?);
 
-    let engine = Engine::new(client.clone()).start();
+    let engine = Engine::new(&config, client.clone()).start();
 
     let server_handle = RpcServer::new(client, engine.clone()).run(&config).await?;