Even more features

Author: macladson

Cargo.lock

@@ -0,0 +1 @@
+config.yaml

watch/.gitignore

@@ -47,12 +47,12 @@ will be able to sync to the beacon node.
 
 1. Run the updater daemon:
 ```
-cargo run -- start-daemon
+cargo run --release -- start-daemon
 ```
 
 1. Start the HTTP API server:
 ```
-cargo run -- serve
+cargo run --release -- serve
 ```
 
 1. Ensure connectivity:
@@ -115,12 +115,6 @@ As beacon.watch continues to develop, more endpoints will be added.
 - More API endpoints
   - E.g. `/v1/proposer_info?start_epoch={}&end_epoch={}`
   - Use new API design.
-  - Finish API for validators/attestations
-
-
-- Fix validator updating.
-  - Only update new validators rather than reloading the whole validator set every update. 
-  - Implement exit epochs (and check for updates)
 
 
 - Store the config in the database on first run so that we can warn against unexpected config changes.

watch/README.md

@@ -1,3 +1,4 @@
+---
 database:
   user: "postgres"
   password: "postgres"
@@ -28,6 +29,16 @@ updater:
   # Whether to sync the block_packing table.
   block_packing: true
 
+blockprint:
+  # Whether to sync client information from blockprint.
+  enabled: false
+  # The URL of the blockprint server.
+  url: "https://api.blockprint.sigp.io"
+  # The username used to authenticate to the blockprint server.
+  username: ""
+  # The password used to authenticate to the blockprint server.
+  password: ""
+
 # Slots per Epoch of the target network.
 slots_per_epoch: 32
 # Log level.

watch/config.yaml.default

@@ -2,7 +2,7 @@ CREATE TABLE validators (
     index integer PRIMARY KEY,
     public_key bytea NOT NULL,
     status text NOT NULL,
-    balance bigint NOT NULL,
-    activation_epoch integer NOT NULL,
+    client text,
+    activation_epoch integer,
     exit_epoch integer
 )

watch/migrations/2022-01-01-000002_validators/up.sql

@@ -1,5 +1,5 @@
 CREATE TABLE suboptimal_attestations (
-    epoch_start_slot integer REFERENCES canonical_slots(slot) ON DELETE CASCADE,
+    epoch_start_slot integer CHECK (epoch_start_slot % 32 = 0) REFERENCES canonical_slots(slot) ON DELETE CASCADE,
     index integer NOT NULL REFERENCES validators(index) ON DELETE CASCADE,
     source boolean NOT NULL,
     head boolean NOT NULL,

watch/migrations/2022-01-01-000013_suboptimal_attestations/up.sql

@@ -0,0 +1,40 @@
+use serde::{Deserialize, Serialize};
+
+pub const fn enabled() -> bool {
+    false
+}
+
+pub const fn url() -> Option<String> {
+    None
+}
+
+pub const fn username() -> Option<String> {
+    None
+}
+
+pub const fn password() -> Option<String> {
+    None
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct Config {
+    #[serde(default = "enabled")]
+    pub enabled: bool,
+    #[serde(default = "url")]
+    pub url: Option<String>,
+    #[serde(default = "username")]
+    pub username: Option<String>,
+    #[serde(default = "password")]
+    pub password: Option<String>,
+}
+
+impl Default for Config {
+    fn default() -> Self {
+        Config {
+            enabled: enabled(),
+            url: url(),
+            username: username(),
+            password: password(),
+        }
+    }
+}

watch/src/blockprint/config.rs

@@ -0,0 +1 @@
+

watch/src/blockprint/error.rs

@@ -0,0 +1,131 @@
+use eth2::SensitiveUrl;
+use reqwest::{Client, Response, Url};
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+use std::time::Duration;
+use types::Slot;
+
+pub use config::Config;
+
+mod config;
+mod error;
+
+const TIMEOUT: Duration = Duration::from_secs(50);
+
+#[derive(Debug)]
+pub enum Error {
+    Reqwest(reqwest::Error),
+    Url(url::ParseError),
+    BlockprintNotSynced,
+    Other(String),
+}
+
+impl From<reqwest::Error> for Error {
+    fn from(e: reqwest::Error) -> Self {
+        Error::Reqwest(e)
+    }
+}
+
+impl From<url::ParseError> for Error {
+    fn from(e: url::ParseError) -> Self {
+        Error::Url(e)
+    }
+}
+
+pub struct WatchBlockprintClient {
+    pub client: Client,
+    pub server: SensitiveUrl,
+    pub username: Option<String>,
+    pub password: Option<String>,
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+pub struct BlockprintSyncingResponse {
+    pub greatest_block_slot: Slot,
+    pub synced: bool,
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+pub struct BlockprintResponse {
+    pub proposer_index: i32,
+    pub slot: Slot,
+    pub best_guess_single: String,
+}
+
+impl WatchBlockprintClient {
+    async fn get(&self, url: Url) -> Result<Response, Error> {
+        let mut builder = self.client.get(url).timeout(TIMEOUT);
+        if let Some(username) = &self.username {
+            builder = builder.basic_auth(username, self.password.as_ref());
+        }
+        let response = builder.send().await.map_err(Error::Reqwest)?;
+
+        if !response.status().is_success() {
+            return Err(Error::Other(response.text().await?));
+        }
+
+        Ok(response)
+    }
+
+    // Returns the `greatest_block_slot` as reported by the Blockprint server.
+    // Will error if the Blockprint server is not synced.
+    pub async fn ensure_synced(&self) -> Result<Slot, Error> {
+        let url = self.server.full.join("sync/")?.join("status")?;
+
+        let response = self.get(url).await?;
+
+        let result = response.json::<BlockprintSyncingResponse>().await?;
+        if !result.synced {
+            return Err(Error::BlockprintNotSynced);
+        }
+
+        Ok(result.greatest_block_slot)
+    }
+
+    pub async fn blockprint_all_validators(
+        &self,
+        highest_validator: i32,
+    ) -> Result<HashMap<i32, String>, Error> {
+        let url = self
+            .server
+            .full
+            .join("validator/")?
+            .join("blocks/")?
+            .join("latest")?;
+
+        let response = self.get(url).await?;
+
+        let mut result = response.json::<Vec<BlockprintResponse>>().await?;
+        result.retain(|print| print.proposer_index <= highest_validator);
+
+        let mut map: HashMap<i32, String> = HashMap::with_capacity(result.len());
+        for print in result {
+            map.insert(print.proposer_index, print.best_guess_single);
+        }
+
+        Ok(map)
+    }
+
+    pub async fn blockprint_proposers_between(
+        &self,
+        start_slot: Slot,
+        end_slot: Slot,
+    ) -> Result<HashMap<i32, String>, Error> {
+        let url = self
+            .server
+            .full
+            .join("blocks/")?
+            .join(&format!("{start_slot}/{end_slot}"))?;
+
+        let response = self.get(url).await?;
+
+        let result = response.json::<Vec<BlockprintResponse>>().await?;
+
+        let mut map: HashMap<i32, String> = HashMap::with_capacity(result.len());
+        for print in result {
+            map.insert(print.proposer_index, print.best_guess_single);
+        }
+
+        Ok(map)
+    }
+}

watch/src/blockprint/mod.rs

@@ -1,3 +1,4 @@
+use crate::blockprint::Config as BlockprintConfig;
 use crate::database::Config as DatabaseConfig;
 use crate::server::Config as ServerConfig;
 use crate::updater::Config as UpdaterConfig;
@@ -16,6 +17,8 @@ fn log_level() -> String {
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct Config {
+    #[serde(default)]
+    pub blockprint: BlockprintConfig,
     #[serde(default)]
     pub database: DatabaseConfig,
     #[serde(default)]
@@ -33,6 +36,7 @@ pub struct Config {
 impl Default for Config {
     fn default() -> Self {
         Self {
+            blockprint: BlockprintConfig::default(),
             database: DatabaseConfig::default(),
             server: ServerConfig::default(),
             updater: UpdaterConfig::default(),