[MTG-972] synchronizer optimized

Makefile

@@ -1,9 +1,10 @@
-.PHONY: build start build-integrity-verification start-integrity-verification start-raw-backfiller start-core-indexing dev stop clippy test
-
+.PHONY: build start build-integrity-verification start-integrity-verification start-core-indexing dev stop clippy test
+#  start-backfiller
 SHELL := /bin/bash
 
 build:
-	@docker compose -f docker-compose.yaml build ingester raw-backfiller das-api synchronizer core-indexing slot-persister
+	@docker compose -f docker-compose.yaml build ingester das-api synchronizer core-indexing slot-persister
+	# backfiller
 
 start:
 	@docker compose -f docker-compose.yaml up -d ingester
@@ -26,8 +27,8 @@ build-integrity-verification:
 start-integrity-verification:
 	@docker compose -f docker-compose.yaml up -d integrity-verification
 
-start-raw-backfiller:
-	@docker compose -f docker-compose.yaml up -d raw-backfiller
+# start-backfiller:
+# 	@docker compose -f docker-compose.yaml up -d backfiller
 
 start-core-indexing:
 	@docker compose -f docker-compose.yaml up -d core-indexing

docker-compose.yaml

@@ -3,7 +3,7 @@ services:
   ingester:
     container_name: ingester
     restart: always
-    entrypoint: sh -c "if [ -z '$$MALLOC_CONF' ]; then exec ./ingester --slots-db-path $$INGESTER_SLOTS_DB_PATH --secondary-slots-db-path $$INGESTER_SECONDARY_SLOTS_DB_PATH; else exec ./profiling_ingester --slots-db-path $$INGESTER_SLOTS_DB_PATH --secondary-slots-db-path $$INGESTER_SECONDARY_SLOTS_DB_PATH; fi"
+    entrypoint: sh -c "if [ -z '${MALLOC_CONF}' ]; then exec ./ingester --slots-db-path $$INGESTER_SLOTS_DB_PATH --secondary-slots-db-path $$INGESTER_SECONDARY_SLOTS_DB_PATH; else exec ./profiling_ingester --slots-db-path $$INGESTER_SLOTS_DB_PATH --secondary-slots-db-path $$INGESTER_SECONDARY_SLOTS_DB_PATH; fi"
     env_file:
       - .env
     network_mode: host
@@ -34,7 +34,7 @@ services:
   das-api:
     container_name: das-api
     restart: always
-    entrypoint: sh -c "if [ -z '$$MALLOC_CONF' ]; then exec ./api; else exec ./profiling_api; fi"
+    entrypoint: sh -c "if [ -z '${MALLOC_CONF}' ]; then exec ./api; else exec ./profiling_api; fi"
     env_file:
       - .env
     network_mode: host
@@ -57,7 +57,7 @@ services:
   synchronizer:
     container_name: synchronizer
     restart: always
-    entrypoint: sh -c "if [ -z '$$MALLOC_CONF' ]; then exec ./synchronizer; else exec ./profiling_synchronizer; fi"
+    entrypoint: sh -c "if [ -z '${MALLOC_CONF}' ]; then exec ./synchronizer; else exec ./profiling_synchronizer; fi"
     env_file:
       - .env
     network_mode: host
@@ -97,53 +97,46 @@ services:
       options:
         max-size: "2048m"
 
-  raw-backfiller:
-    container_name: raw-backfiller
-    restart: always
-    entrypoint: sh -c "if [ -z '$$MALLOC_CONF' ]; then exec ./raw_backfiller; else exec ./profiling_raw_backfiller; fi"
-    env_file:
-      - .env
-    network_mode: host
-    volumes:
-      - ${INGESTER_ROCKS_DB_PATH}:${INGESTER_ROCKS_DB_PATH_CONTAINER}:rw
-      - ${INGESTER_PROFILING_FILE_PATH}:${INGESTER_PROFILING_FILE_PATH_CONTAINER}:rw
-      - ./creds.json:/usr/src/app/creds.json
-      - ./heaps:/usr/src/app/heaps:rw
-    stop_grace_period: 5m
-    build:
-      context: .
-      dockerfile: ingester.Dockerfile
-    logging:
-      options:
-        max-size: "2048m"
+  # TODO: pass the arguments to the container, check the mounts as we have 2 dbs - slots and the main one
+  # backfill:
+  #   container_name: backfill
+  #   restart: always
+  #   entrypoint: sh -c "if [ -z '$$MALLOC_CONF' ]; then exec ./backfill; else exec ./profiling_backfill; fi"
+  #   env_file:
+  #     - .env
+  #   network_mode: host
+  #   volumes:
+  #     - ${INGESTER_ROCKS_DB_PATH}:${INGESTER_ROCKS_DB_PATH_CONTAINER}:rw
+  #     - ${INGESTER_PROFILING_FILE_PATH}:${INGESTER_PROFILING_FILE_PATH_CONTAINER}:rw
+  #     - ./creds.json:/usr/src/app/creds.json
+  #     - ./heaps:/usr/src/app/heaps:rw
+  #   stop_grace_period: 5m
+  #   build:
+  #     context: .
+  #     dockerfile: ingester.Dockerfile
+  #   logging:
+  #     options:
+  #       max-size: "2048m"
 
   slot-persister:
     container_name: slot-persister
     restart: always
-    entrypoint: |
-      sh -c "
-      ARGS=\"--target-db-path $target_db_path\"
-      ARGS=\"$$ARGS --rpc-host $rpc_host\"
-      [ -n \"$start_slot\" ] && ARGS=\"$$ARGS --start-slot $start_slot\"
-      [ -n \"$big_table_credentials\" ] && ARGS=\"$$ARGS --big-table-credentials $big_table_credentials\"
-      [ -n \"$big_table_timeout\" ] && ARGS=\"$$ARGS --big-table-timeout $big_table_timeout\"
-      [ -n \"$metrics_port\" ] && ARGS=\"$$ARGS --metrics-port $metrics_port\"
-      [ -n \"$chunk_size\" ] && ARGS=\"$$ARGS --chunk-size $chunk_size\"
-      [ -n \"$max_concurrency\" ] && ARGS=\"$$ARGS --max-concurrency $max_concurrency\"
-
-      if [ -z \"$MALLOC_CONF\" ]; then
-        exec ./slot_persister $$ARGS
-      else
-        exec ./profiling_slot_persister $$ARGS
-      fi"
+    entrypoint: >
+      sh -c "if [ -f /usr/src/app/creds.json ]; then
+        export BIG_TABLE_CREDENTIALS=/usr/src/app/creds.json;
+      fi;
+      exec ./slot_persister"
     env_file:
       - .env
+    environment:
+      SLOTS_DB_PRIMARY_PATH: ${INGESTER_SLOTS_DB_PATH}
+      RPC_HOST: ${INGESTER_RPC_HOST}
+      # BIG_TABLE_CREDENTIALS: /usr/src/app/creds.json # refactored this to account for the file doesn't exist case
+      METRICS_PORT: 9090
     network_mode: host
     volumes:
-      - ${target_db_path}:${target_db_path}:rw
-      - ${INGESTER_PROFILING_FILE_PATH}:${INGESTER_PROFILING_FILE_PATH_CONTAINER}:rw
-      - ${big_table_credentials:-/tmp/creds.json}:${big_table_credentials:-/tmp/creds.json}
-      - ./heaps:/usr/src/app/heaps:rw
+      - ${INGESTER_SLOTS_DB_PATH}:${INGESTER_SLOTS_DB_PATH}:rw
+      - ./creds.json:/usr/src/app/creds.json
     stop_grace_period: 5m
     build:
       context: .

ingester.Dockerfile

@@ -18,6 +18,7 @@ COPY rocks-db ./rocks-db
 COPY tests/setup ./tests/setup
 COPY usecase ./usecase
 COPY integrity_verification ./integrity_verification
+COPY integration_tests ./integration_tests
 
 RUN cargo chef prepare --recipe-path recipe.json
 
@@ -36,12 +37,12 @@ RUN cargo chef cook --release --recipe-path recipe.json
 # Building the services
 FROM cacher AS builder
 COPY . .
-RUN cargo build --release --bin ingester --bin api --bin raw_backfiller --bin synchronizer --bin slot_persister
+RUN cargo build --release --bin ingester --bin api --bin backfill --bin synchronizer --bin slot_persister
 
 # Building the profiling feature services
 FROM cacher AS builder-with-profiling
 COPY . .
-RUN cargo build --release --features profiling --bin ingester --bin api --bin raw_backfiller --bin synchronizer --bin slot_persister
+RUN cargo build --release --features profiling --bin ingester --bin api --bin backfill --bin synchronizer --bin slot_persister
 
 # Final image
 FROM rust:1.76-slim-bullseye AS runtime
@@ -52,12 +53,12 @@ ENV TZ=Etc/UTC APP_USER=appuser LD_PRELOAD="/usr/local/lib/libjemalloc.so.2"
 RUN groupadd $APP_USER && useradd -g $APP_USER $APP_USER && mkdir -p ${APP}
 
 COPY --from=builder /rust/target/release/ingester ${APP}/ingester
-COPY --from=builder /rust/target/release/raw_backfiller ${APP}/raw_backfiller
+COPY --from=builder /rust/target/release/backfill ${APP}/backfill
 COPY --from=builder /rust/target/release/api ${APP}/api
 COPY --from=builder /rust/target/release/synchronizer ${APP}/synchronizer
 COPY --from=builder /rust/target/release/slot_persister ${APP}/slot_persister
 COPY --from=builder-with-profiling /rust/target/release/ingester ${APP}/profiling_ingester
-COPY --from=builder-with-profiling /rust/target/release/raw_backfiller ${APP}/profiling_raw_backfiller
+COPY --from=builder-with-profiling /rust/target/release/backfill ${APP}/profiling_backfill
 COPY --from=builder-with-profiling /rust/target/release/api ${APP}/profiling_api
 COPY --from=builder-with-profiling /rust/target/release/synchronizer ${APP}/profiling_synchronizer
 COPY --from=builder-with-profiling /rust/target/release/slot_persister ${APP}/profiling_slot_persister

integration_tests/src/common.rs

@@ -35,7 +35,6 @@ use tokio::{
 };
 use usecase::proofs::MaybeProofChecker;
 
-use tracing::{error, info};
 use serde::de::DeserializeOwned;
 use solana_account_decoder::{UiAccount, UiAccountEncoding};
 use solana_client::{
@@ -51,6 +50,7 @@ use solana_sdk::{
 };
 use solana_transaction_status::{EncodedConfirmedTransactionWithStatusMeta, UiTransactionEncoding};
 use std::{fmt, time::Duration};
+use tracing::{error, info};
 
 use std::path::PathBuf;
 

interface/src/price_fetcher.rs

@@ -1,6 +1,5 @@
 use crate::error::UsecaseError;
 use async_trait::async_trait;
-use solana_program::pubkey::Pubkey;
 use std::collections::HashMap;
 
 #[async_trait]

nft_ingester/Cargo.toml

@@ -33,7 +33,7 @@ stretto = { workspace = true }
 async-channel = { workspace = true }
 tokio-util = { workspace = true }
 tracing-subscriber = { workspace = true }
-clap = { workspace = true }
+clap = { workspace = true, features = ["env"] }
 bincode = { workspace = true }
 metrics-utils = { path = "../metrics_utils" }
 rocks-db = { path = "../rocks-db" }
@@ -84,6 +84,7 @@ indicatif = { workspace = true }
 tokio-retry = { workspace = true }
 axum = { workspace = true }
 rocksdb = { workspace = true }
+num-bigint = "0.4"
 
 [dev-dependencies]
 setup = { path = "../tests/setup" }
@@ -145,3 +146,6 @@ name = "explorer"
 
 [[bin]]
 name = "synchronizer_utils"
+
+[[bin]]
+name = "dumper"

nft_ingester/src/api/dapi/asset.rs

@@ -18,7 +18,7 @@ use interface::processing_possibility::ProcessingPossibilityChecker;
 use itertools::Itertools;
 use metrics_utils::ApiMetricsConfig;
 use rocks_db::asset::{AssetLeaf, AssetSelectedMaps};
-use rocks_db::{AssetAuthority, Storage};
+use rocks_db::Storage;
 use tokio::sync::Mutex;
 use tokio::task::{JoinError, JoinSet};
 
@@ -140,7 +140,9 @@ fn asset_selected_maps_into_full_asset(
             if let Some(asset_static_details) = &asset_complete_details.static_details {
                 // collection itself cannot have a collection
                 // TODO!: should we also include in this check FungibleToken?
-                if &asset_static_details.specification_asset_class != &SpecificationAssetClass::MplCoreCollection {
+                if &asset_static_details.specification_asset_class
+                    != &SpecificationAssetClass::MplCoreCollection
+                {
                     if let Some(collection_details) = &asset_complete_details.collection {
                         if !collection_details.is_collection_verified.value {
                             return None;
@@ -294,7 +296,7 @@ pub async fn get_by_ids<
                             },
                         );
                     }
-                    Err(e) => {}
+                    Err(_) => {}
                 }
             }
 

nft_ingester/src/api/dapi/rpc_asset_convertors.rs

@@ -279,7 +279,12 @@ pub fn to_authority(
     // even if there is no authority for asset we should not set Pubkey::default(), just empty string
     let auth_key = update_authority
         .map(|update_authority| update_authority.to_string())
-        .unwrap_or(authority.as_ref().map(|auth| auth.authority.to_string()).unwrap_or("".to_string()));
+        .unwrap_or(
+            authority
+                .as_ref()
+                .map(|auth| auth.authority.to_string())
+                .unwrap_or("".to_string()),
+        );
 
     vec![Authority {
         address: auth_key,

nft_ingester/src/backfiller.rs

@@ -1,4 +1,4 @@
-use crate::config::{BackfillerConfig, BackfillerSourceMode, BigTableConfig};
+use crate::config::{BackfillerSourceMode, BigTableConfig};
 use crate::error::IngesterError;
 use async_trait::async_trait;
 use backfill_rpc::rpc::BackfillRPC;
@@ -118,7 +118,6 @@ impl BlockProducer for BackfillSource {
 
 #[derive(Clone)]
 pub struct TransactionsParser<C: BlockConsumer, P: BlockProducer, S: SlotGetter> {
-    rocks_client: Arc<rocks_db::Storage>,
     slot_getter: Arc<S>,
     consumer: Arc<C>,
     producer: Arc<P>,
@@ -134,7 +133,6 @@ where
     S: SlotGetter,
 {
     pub fn new(
-        rocks_client: Arc<rocks_db::Storage>,
         slot_getter: Arc<S>,
         consumer: Arc<C>,
         producer: Arc<P>,
@@ -143,7 +141,6 @@ where
         chunk_size: usize,
     ) -> TransactionsParser<C, P, S> {
         TransactionsParser {
-            rocks_client,
             slot_getter,
             consumer,
             producer,

nft_ingester/src/bin/api/main.rs

@@ -30,10 +30,10 @@ pub const DEFAULT_SECONDARY_ROCKSDB_PATH: &str = "./my_rocksdb_secondary";
 
 #[tokio::main(flavor = "multi_thread")]
 pub async fn main() -> Result<(), IngesterError> {
+    tracing_subscriber::fmt::init();
     info!("Starting API server...");
-
     let config: ApiConfig = setup_config("API_");
-    init_logger(&config.get_log_level());
+    // init_logger(&config.get_log_level());
 
     let guard = if config.run_profiling {
         Some(
@@ -73,6 +73,7 @@ pub async fn main() -> Result<(), IngesterError> {
         config.database_config.get_database_url()?.as_str(),
         min_connections,
         max_connections,
+        None,
         red_metrics.clone(),
     )
     .await?;
@@ -120,6 +121,7 @@ pub async fn main() -> Result<(), IngesterError> {
                         rocks_storage.clone(),
                         json_downloader_metrics.clone(),
                         red_metrics.clone(),
+                        nft_ingester::config::default_parallel_json_downloaders(),
                     )
                     .await,
                 ))