[STATS] Marking approximate points + general improvements

stats/.dockerignore

@@ -3,3 +3,4 @@ Dockerfile
 README.md
 tests
 config.toml
+data

stats/.gitignore

@@ -0,0 +1 @@
+data/

stats/docker-compose.dev.yml

@@ -0,0 +1,122 @@
+version: '3.9'
+
+services:
+  db-init:
+    image: postgres:15
+    volumes:
+      - ./data/blockscout-db:/var/lib/postgresql/data
+    entrypoint:
+      - sh
+      - -c
+      - |
+        chown -R 2000:2000 /var/lib/postgresql/data
+
+  db:
+    depends_on:
+      db-init:
+        condition: service_completed_successfully
+    image: postgres:15
+    user: 2000:2000
+    shm_size: 256m
+    restart: always
+    container_name: 'db'
+    command: postgres -c 'max_connections=200' -c 'client_connection_check_interval=60000'
+    environment:
+      POSTGRES_DB: 'blockscout'
+      POSTGRES_USER: 'blockscout'
+      POSTGRES_PASSWORD: 'ceWb1MeLBEeOIfk65gU8EjF8'
+    ports:
+      - target: 5432
+        published: 7432
+    volumes:
+      - ./data/blockscout-db:/var/lib/postgresql/data
+    healthcheck:
+      test: [ "CMD-SHELL", "pg_isready -U blockscout -d blockscout" ]
+      interval: 10s
+      timeout: 5s
+      retries: 5
+      start_period: 10s
+
+  backend:
+    depends_on:
+      - db
+    image: blockscout/blockscout:6.4.0
+    links:
+      - db:database
+    # extra_hosts:
+    #   - 'host.docker.internal:host-gateway'
+    environment:
+      DATABASE_URL: postgresql://blockscout:ceWb1MeLBEeOIfk65gU8EjF8@db:5432/blockscout
+      ETHEREUM_JSONRPC_VARIANT: erigon
+      ETHEREUM_JSONRPC_HTTP_URL: http://host.docker.internal:8545/
+      ETHEREUM_JSONRPC_TRACE_URL: http://host.docker.internal:8545/
+      FIRST_BLOCK: 5660029
+      ECTO_USE_SSL: false
+      PORT: 4000
+    ports:
+      - 80:4000
+    command:
+      [
+        "/bin/sh",
+        "-c",
+        "bin/blockscout eval \"Elixir.Explorer.ReleaseTasks.create_and_migrate()\" && bin/blockscout start"
+      ]
+
+  stats-db-init:
+    image: postgres:15
+    volumes:
+      - ./data/stats-db:/var/lib/postgresql/data
+    entrypoint:
+      - sh
+      - -c
+      - |
+        chown -R 2000:2000 /var/lib/postgresql/data
+
+  stats-db:
+    depends_on:
+      stats-db-init:
+        condition: service_completed_successfully
+    image: postgres:15
+    user: 2000:2000
+    shm_size: 256m
+    restart: always
+    container_name: 'stats-db'
+    command: postgres -c 'max_connections=200'
+    environment:
+      POSTGRES_DB: 'stats'
+      POSTGRES_USER: 'stats'
+      POSTGRES_PASSWORD: 'n0uejXPl61ci6ldCuE2gQU5Y'
+    ports:
+      - target: 5432
+        published: 7433
+    volumes:
+      - ./data/stats-db:/var/lib/postgresql/data
+    healthcheck:
+      test: [ "CMD-SHELL", "pg_isready -U stats -d stats" ]
+      interval: 10s
+      timeout: 5s
+      retries: 5
+      start_period: 10s
+
+  stats:
+    depends_on:
+      - stats-db
+      - backend
+    build: .
+    pull_policy: always
+    platform: linux/amd64
+    restart: always
+    container_name: 'stats'
+    # extra_hosts:
+    #   - 'host.docker.internal:host-gateway'
+    environment:
+      - STATS__DB_URL=postgres://stats:n0uejXPl61ci6ldCuE2gQU5Y@stats-db:5432/stats
+      - STATS__BLOCKSCOUT_DB_URL=${STATS__BLOCKSCOUT_DB_URL:-postgresql://blockscout:ceWb1MeLBEeOIfk65gU8EjF8@db:5432/blockscout}
+      - STATS__CREATE_DATABASE=true
+      - STATS__RUN_MIGRATIONS=true
+      - STATS__SERVER__HTTP__ENABLED=true
+      - STATS__SERVER__HTTP__ADDR=0.0.0.0:8050
+      - STATS__SERVER__HTTP__MAX_BODY_SIZE=2097152
+      - RUST_BACKTRACE=1
+    ports:
+      - 8080:8050

stats/justfile

@@ -25,6 +25,8 @@ test *args:
     cargo test {{args}} -- --include-ignored
 
 test-with-db *args:
+    # remove db from previous run (if failed)
+    -just                            docker-name="{{docker-name}}-test" stop-postgres 2> /dev/null
     -just db-port="{{test-db-port}}" db-name="" docker-name="{{docker-name}}-test" start-postgres
     just db-port="{{test-db-port}}" db-name=""                                    test {{args}}
     just                            docker-name="{{docker-name}}-test" stop-postgres

stats/stats-proto/Cargo.toml

@@ -13,6 +13,9 @@ serde = { version = "1", features = ["derive"] }
 serde_with = { version = "2.0", features = ["hex", "base64"] }
 async-trait = "0.1"
 
+[dev-dependencies]
+serde_json = "1.0"
+
 [build-dependencies]
 actix-prost-build = { git = "https://github.com/blockscout/actix-prost" }
 tonic-build = "0.8"

stats/stats-proto/build.rs

@@ -16,7 +16,9 @@ fn compile(
         .protoc_arg("--openapiv2_opt")
         .protoc_arg("grpc_api_configuration=proto/api_config_http.yaml,output_format=yaml,allow_merge=true,merge_file_name=stats")
         .bytes(["."])
-        .type_attribute(".", "#[actix_prost_macros::serde]");
+        .type_attribute(".", "#[actix_prost_macros::serde]")
+        .field_attribute(".blockscout.stats.v1.Point.is_approximate", "#[serde(skip_serializing_if = \"std::ops::Not::not\")]")
+        .field_attribute(".blockscout.stats.v1.Point.is_approximate", "#[serde(default)]");
 
     config.compile_protos(protos, includes)?;
     Ok(())

stats/stats-proto/proto/stats.proto

@@ -34,6 +34,7 @@ message GetLineChartRequest {
 message Point {
   string date = 1;
   string value = 2;
+  bool is_approximate = 3;
 }
 
 message LineChart { repeated Point chart = 1; }
@@ -53,4 +54,4 @@ message LineChartSection {
   repeated LineChartInfo charts = 3;
 }
 
-message LineCharts { repeated LineChartSection sections = 1; }
+message LineCharts { repeated LineChartSection sections = 1; }

stats/stats-proto/src/lib.rs

@@ -6,3 +6,6 @@ pub mod blockscout {
         }
     }
 }
+
+#[cfg(test)]
+mod tests;

stats/stats-proto/src/tests.rs

@@ -0,0 +1,39 @@
+use prost::Message;
+
+use crate::blockscout::stats::v1::{self as proto};
+
+const PRECISE_POINT_1: &str = r#"
+{
+    "date": "2024-03-14",
+    "value": "188542399",
+    "isApproximate": false
+}
+"#;
+
+const PRECISE_POINT_2: &str = r#"
+{
+    "date": "2024-03-14",
+    "value": "188542399"
+}
+"#;
+
+#[test]
+fn is_approximate_serialization() {
+    // deserialize
+    let point: proto::Point = serde_json::from_str(PRECISE_POINT_1).unwrap();
+    assert!(!point.is_approximate);
+    let point: proto::Point = serde_json::from_str(PRECISE_POINT_2).unwrap();
+    assert!(!point.is_approximate);
+
+    // serialize
+    let point = proto::Point {
+        date: "2024-03-14".to_owned(),
+        value: "188542399".to_owned(),
+        is_approximate: false,
+    };
+    let serialized_point = serde_json::to_string(&point).unwrap();
+    assert_eq!(
+        serialized_point.replace([' ', '\n'], ""),
+        PRECISE_POINT_2.replace([' ', '\n'], "")
+    );
+}

stats/stats-proto/swagger/stats.swagger.yaml

@@ -190,4 +190,6 @@ definitions:
         type: string
       value:
         type: string
+      isApproximate:
+        type: boolean
     title: All integers are encoded as strings to prevent data loss

stats/stats-server/src/charts.rs

@@ -2,17 +2,17 @@ use crate::config::{
     toml_config::{Config, LineChartSection},
     ChartSettings,
 };
-use stats::{cache::Cache, counters, entity::sea_orm_active_enums::ChartType, lines, Chart};
+use stats::{cache::Cache, counters, entity::sea_orm_active_enums::ChartType, lines, ChartUpdater};
 use std::{
     collections::{BTreeMap, HashMap, HashSet},
     hash::Hash,
     sync::Arc,
 };
 
-pub type ArcChart = Arc<dyn Chart + Send + Sync + 'static>;
+pub type ArcChartUpdater = Arc<dyn ChartUpdater + Send + Sync + 'static>;
 
 pub struct ChartInfo {
-    pub chart: ArcChart,
+    pub chart: ArcChartUpdater,
     pub settings: ChartSettings,
 }
 
@@ -131,7 +131,7 @@ impl Charts {
             .collect()
     }
 
-    fn all_charts() -> Vec<ArcChart> {
+    fn all_charts() -> Vec<ArcChartUpdater> {
         let accounts_cache = Cache::default();
         let new_txns = Arc::new(lines::NewTxns::default());
         let new_native_coin_transfers = Arc::new(lines::NewNativeCoinTransfers::default());

stats/stats-server/src/read_service.rs

@@ -1,6 +1,6 @@
-use crate::{charts::Charts, serializers::serialize_line_points};
+use crate::{charts::Charts, serializers::serialize_line_points, settings::LimitsSettings};
 use async_trait::async_trait;
-use chrono::NaiveDate;
+use chrono::{Duration, NaiveDate, Utc};
 use sea_orm::{DatabaseConnection, DbErr};
 use stats::ReadError;
 use stats_proto::blockscout::stats::v1::{
@@ -14,17 +14,37 @@ use tonic::{Request, Response, Status};
 pub struct ReadService {
     db: Arc<DatabaseConnection>,
     charts: Arc<Charts>,
+    limits: ReadLimits,
 }
 
 impl ReadService {
-    pub async fn new(db: Arc<DatabaseConnection>, charts: Arc<Charts>) -> Result<Self, DbErr> {
-        Ok(Self { db, charts })
+    pub async fn new(
+        db: Arc<DatabaseConnection>,
+        charts: Arc<Charts>,
+        limits: ReadLimits,
+    ) -> Result<Self, DbErr> {
+        Ok(Self { db, charts, limits })
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ReadLimits {
+    /// See [`LimitsSettings::request_interval_limit_days`]
+    pub request_interval_limit: Duration,
+}
+
+impl From<LimitsSettings> for ReadLimits {
+    fn from(value: LimitsSettings) -> Self {
+        Self {
+            request_interval_limit: Duration::days(value.request_interval_limit_days.into()),
+        }
     }
 }
 
 fn map_read_error(err: ReadError) -> Status {
     match &err {
         ReadError::NotFound(_) => Status::not_found(err.to_string()),
+        ReadError::IntervalLimitExceeded(_) => Status::invalid_argument(err.to_string()),
         _ => {
             tracing::error!(err = ?err, "internal read error");
             Status::internal(err.to_string())
@@ -56,7 +76,7 @@ impl StatsService for ReadService {
             .filter_map(|(counter, info)| {
                 data.remove(&counter.id).map(|point| {
                     let point: stats::DateValue = if info.chart.relevant_or_zero() {
-                        point.relevant_or_zero()
+                        point.relevant_or_zero(Utc::now().date_naive())
                     } else {
                         point
                     };
@@ -90,18 +110,20 @@ impl StatsService for ReadService {
             .and_then(|date| NaiveDate::from_str(&date).ok());
         let to = request.to.and_then(|date| NaiveDate::from_str(&date).ok());
         let policy = Some(chart_info.chart.missing_date_policy());
-        let mut data = stats::get_chart_data(&self.db, &request.name, from, to, policy)
-            .await
-            .map_err(map_read_error)?;
+        let mark_approx = chart_info.chart.approximate_trailing_points();
+        let interval_limit = Some(self.limits.request_interval_limit);
+        let data = stats::get_chart_data(
+            &self.db,
+            &request.name,
+            from,
+            to,
+            interval_limit,
+            policy,
+            mark_approx,
+        )
+        .await
+        .map_err(map_read_error)?;
 
-        if chart_info.chart.drop_last_point() {
-            // remove last data point, because it can be partially updated
-            if let Some(last) = data.last() {
-                if last.is_partial() {
-                    data.pop();
-                }
-            }
-        }
         let serialized_chart = serialize_line_points(data);
         Ok(Response::new(LineChart {
             chart: serialized_chart,

stats/stats-server/src/serializers.rs

@@ -1,11 +1,12 @@
-use stats::DateValue;
+use stats::ExtendedDateValue;
 use stats_proto::blockscout::stats::v1::Point;
 
-pub fn serialize_line_points(data: Vec<DateValue>) -> Vec<Point> {
+pub fn serialize_line_points(data: Vec<ExtendedDateValue>) -> Vec<Point> {
     data.into_iter()
         .map(|point| Point {
             date: point.date.to_string(),
             value: point.value,
+            is_approximate: point.is_approximate,
         })
         .collect()
 }

stats/stats-server/src/server.rs

@@ -78,7 +78,7 @@ pub async fn stats(settings: Settings) -> Result<(), anyhow::Error> {
             .await;
     });
 
-    let read_service = Arc::new(ReadService::new(db, charts).await?);
+    let read_service = Arc::new(ReadService::new(db, charts, settings.limits.into()).await?);
     let health = Arc::new(HealthService::default());
 
     let grpc_router = grpc_router(read_service.clone(), health.clone());