Update RPC to use valid hash/height (#126)

Author: calummoore

helm/values.yaml

@@ -41,7 +41,7 @@ extraEnvVarsSecret: ""
 
 ## @param replicaCount Number of Polybase replicas
 ##
-replicaCount: 1
+replicaCount: 3
 ## @param updateStrategy.type updateStrategy for Polybase statefulset
 ## ref: https://kubernetes.io/docs/concepts/workloads/controllers/statefulset/#update-strategies
 ##

polybase/src/db.rs

@@ -1,6 +1,5 @@
 use crate::hash;
 use crate::mempool::Mempool;
-use crate::rollup::Rollup;
 use crate::txn::{self, CallTxn};
 use crate::util;
 use futures::TryStreamExt;
@@ -87,8 +86,6 @@ impl Default for DbConfig {
 pub struct Db {
     mempool: Mempool<[u8; 32], CallTxn, usize, [u8; 32]>,
     gateway: Gateway,
-    // TODO: remove this pub
-    pub rollup: Rollup,
     indexer: Indexer,
     sender: AsyncMutex<mpsc::Sender<CallTxn>>,
     receiver: AsyncMutex<mpsc::Receiver<CallTxn>>,
@@ -107,7 +104,6 @@ impl Db {
 
         Ok(Self {
             mempool: Mempool::new(),
-            rollup: Rollup::new(),
             gateway: gateway::initialize(logger.clone()),
             indexer,
             sender: AsyncMutex::new(sender),

polybase/src/errors/http.rs

@@ -6,7 +6,6 @@ use serde::Serialize;
 use std::{error::Error, fmt::Display};
 
 use super::reason::ReasonCode;
-use crate::rollup;
 use crate::{
     auth,
     db::{self},
@@ -185,9 +184,3 @@ impl From<auth::AuthUserError> for HTTPError {
         HTTPError::new(ReasonCode::from_auth_error(&err), Some(Box::new(err)))
     }
 }
-
-impl From<rollup::RollupError> for HTTPError {
-    fn from(err: rollup::RollupError) -> Self {
-        HTTPError::new(ReasonCode::Internal, Some(Box::new(err)))
-    }
-}

polybase/src/main.rs

@@ -13,7 +13,6 @@ mod errors;
 mod hash;
 mod mempool;
 mod network;
-mod rollup;
 mod rpc;
 mod txn;
 mod util;
@@ -258,7 +257,7 @@ async fn main() -> Result<()> {
         let mut last_commit = Instant::now();
 
         // For migration only, check if DB is empty
-        if db.is_empty().await.unwrap_or(true) {
+        if db.is_empty().await.unwrap_or(true) && solid.height() == 0 {
             info!(logger, "DB is empty, requesting snapshot");
             network
                 .send_all(NetworkEvent::SnapshotRequest {
@@ -337,8 +336,6 @@ async fn main() -> Result<()> {
                         // We've been offered a snapshot from another peer, we should accept this offer if we
                         // don't already have an ongoing snapshot in progress
                         NetworkEvent::SnapshotOffer{ id } => {
-                            info!(logger, "Peer offered snapshot, sending accept"; "peer_id" => peer_id.prefix());
-
                             if snapshot_from.is_some() {
                                 debug!(logger, "Already have snapshot in progress, ignoring offer"; "peer_id" => peer_id.prefix(),  "id" => id);
                                 continue;
@@ -349,6 +346,8 @@ async fn main() -> Result<()> {
                                 continue;
                             }
 
+                            info!(logger, "Peer offered snapshot, sending accept"; "peer_id" => peer_id.prefix(), "id" => id);
+
                             // Save who the snapshot is from
                             snapshot_from = Some((network_peer_id.clone(), id));
 
@@ -428,20 +427,21 @@ async fn main() -> Result<()> {
                                 #[allow(clippy::unwrap_used)]
                                 db.restore_chunk(chunk).unwrap();
                             } else {
-                                // Reset snapshot from
-                                snapshot_from = None;
-
                                 // We are finished, reset solid with the new proposal state from the snapshot
                                 #[allow(clippy::unwrap_used)]
                                 let manifest = db.get_manifest().await.unwrap().unwrap();
+                                let height = manifest.height;
                                 solid.reset(manifest);
 
-                                info!(logger, "Restore db from snapshot complete");
+                                // Reset snapshot from
+                                snapshot_from = None;
+
+                                info!(logger, "Restore db from snapshot complete"; "height" => height);
                             }
                         }
 
                         NetworkEvent::Accept { accept } => {
-                            info!(logger, "Received accept"; "height" => &accept.height, "skips" => &accept.skips, "from" => &accept.leader_id.prefix(), "hash" => accept.proposal_hash.to_string());
+                            info!(logger, "Received accept"; "height" => &accept.height, "skips" => &accept.skips, "from" => &accept.leader_id.prefix(), "hash" => accept.proposal_hash.to_string(), "local_height" => solid.height());
                             solid.receive_accept(&accept, &peer_id);
                         }
 

polybase/src/rollup.rs

@@ -413,17 +413,23 @@ async fn health(state: web::Data<RouteState>) -> impl Responder {
 struct StatusResponse {
     status: String,
     root: String,
+    height: usize,
     peers: usize,
-    leader: usize,
 }
 
 #[get("/v0/status")]
 async fn status(state: web::Data<RouteState>) -> Result<web::Json<StatusResponse>, HTTPError> {
+    let manifest = state.db.get_manifest().await?;
+    let height = manifest.as_ref().map(|m| m.height).unwrap_or(0);
+    let hash = manifest
+        .as_ref()
+        .map(|m| m.hash().to_string())
+        .unwrap_or("0x0".to_string());
     Ok(web::Json(StatusResponse {
         status: "OK".to_string(),
-        root: hex::encode(state.db.rollup.root()?),
+        root: hash,
+        height,
         peers: 23,
-        leader: 12,
     }))
 }
 

polybase/src/rpc.rs

@@ -109,6 +109,7 @@ impl Solid {
         );
         let mut events = self.shared.events.lock();
         events.clear();
+        self.reset_skip_timeout();
     }
 
     pub fn run(&self) -> tokio::task::JoinHandle<()> {