RRP-318: Add worker authentication (#11)

* gitignore localtemp

* PoolTest Conf

* ShareLog init

* External Hello Init

* shares_logger init

* log_share Share hash

* static LOGGER_SENDER Bounded channel

* primary_rx backup_rx

* ShareLogProto

* pub struct ShareLog tested

* ShareLogProto removed

* imports remove

* ClickHouse init tested

* insert_batch clickhouse 0.13.1

* share_processor.rs Tested

* with share_diff

* TEMP remove clickhouse channel

* 169-3-1 DebugLog Proto

* 169-3-1 Debugger-Logger-v.1

* services::debug_log commented

* MATERIALIZED VIEW mv_hash_rate_stats

* shares_per_minute to 10

* Backtrace::force_capture

* 169-3-1 ExtendedChannelKind

* rm -rf hello :)

* 169-3-1 new config.rs

* 169-3-1 comments to English

* 169-3-1 README

* 169-3-1 README

* 169-3-1 rm handshake )

* 169-5 ShareLog to models

* ClickhouseShare to models

* 169-5 share to share_log

* TEMP 169-5 New Config Mod (ENV in progress)

* 169-5 TEMP ENV

* 169-5 ENV to settings

* 169-5 README

* 169-5 Syntax Fix

* 169-5 Spaces Fix

* 169-5 ClickhouseError to mod

* 169-5 is_valid to ShareStatus

* 169-5 ClickHouse to trait

* 169-5 Deps Removed

* 169-5 ShareLogger refactoring

* test_highload_share_processing

* 169-5 test_highload_share_processing improve

* 169-5 README Refreshed

* 169-6 ShareData

* 169-6 test_share_types

* 169-6 TEMP BlockFound

* 169-6 shares_logger::log_block(block)

* 169-6 store_share on BlockFound

* 169-7 Queries To Files

* 169-7 Small Fix

* 169-7 README

* 169-7 Cargo Fix

* 169-7 share.is_block_found fn

* 169-7 Small Fixes

* min_individual_miner_hashrate

* RRP-318 nomium_test_worker from Translator to Pool hardcoded

* RRP-318 user_identity from worker

* RRP-318 user_identity to prepare_share_log

* RRP-318 user_identity to ClickHouse 1

* RRP-318 user_identity DB optimized

* RRP-318 user_identity on BlockFound

* 169-11 handle_authorize with reqwest proto

* 169-11 worker_name_store MOD

* 169-11 worker_name_store to shares-logger

* 169-11 shares with worker_id

* 169-11 worker_id on blocks

* 169-11 Tables n MV updated

* 169-11 mv_hash_rate_stats Readme-Ru

* Update rrp-translator.Dockerfile

* Update rrp-translator.Dockerfile

* Update rrp-translator.Dockerfile

* Update rrp-translator.Dockerfile

* Update rrp-translator.Dockerfile

* Update rrp-translator.Dockerfile

* Update rrp-translator.Dockerfile

* Update rrp-translator.Dockerfile

* Update rrp-translator.Dockerfile

* Update rrp-translator.Dockerfile

* Update rrp-translator.Dockerfile

* 169-11 wo OpenSSL

* Update rrp-translator.Dockerfile

* Update docker-compose.yml

* Update docker-compose.yml

* 169-11 new mv_hash_rate_stats

* 169-11 Remove Obsolete Tests and fn to_storage_format

* 169-11 README-RU.md mod

* 169-11 share_data Trait rm

* 169-11 ENV REDROCK_API_URL REDROCK_API_KEY

* 169-11 New MV n HR Algo

* 169-11 small

* 169-11 Timeout to ENV etc

---------

Co-authored-by: Roman Dukhanin <romandukhanin@yandex.ru>
Co-authored-by: inha1e <78920981+inha1e@users.noreply.github.com>

Author: 3 people

cicd/docker-compose.yml

@@ -7,6 +7,9 @@ services:
           - CLICKHOUSE_USER=${CLICKHOUSE_USER}
           - CLICKHOUSE_PASSWORD=${CLICKHOUSE_PASSWORD}
         restart: always
+        privileged: true
+        volumes:
+          - /var/log/clickhouse-server:/var/log/clickhouse-server
         ports:
           - 8123:8123
         network_mode: host

cicd/rrp-translator.Dockerfile

@@ -1,6 +1,7 @@
 FROM rust:1.82 AS build
 
-RUN apt-get update && apt-get install -y musl-tools
+RUN apt-get update 
+RUN apt-get install -y musl-tools
 
 WORKDIR /stratum
 COPY ./ .

nomium/shares-logger/Cargo.toml

@@ -16,6 +16,7 @@ serde_json = "1.0"
 primitive-types = "0.12"
 lazy_static = "1.4"
 config = "0.13"
+parking_lot = "0.12"
 
 [dev-dependencies]
 tokio = { version = "1.0", features = ["rt", "macros", "test-util", "time"] }

nomium/shares-logger/README-RU.md

@@ -0,0 +1,29 @@
+# Инструкция как из консоли Linux пересоздать MV и заполнить его данными из источника
+
+1. Добавить в MV ключевое слово: POPULATE (уже добавил в nomium/shares-logger/src/storage/clickhouse/queries/hashrate_view.sql)
+
+2. Данные доступа и пути к файлам поменять на свои:
+
+Проверка существования представления:
+```bash
+echo "SHOW TABLES FROM mining LIKE 'mv_hash_rate%'" | curl 'http://localhost:8123/?database=mining' --data-binary @- -u default:5555
+```
+
+Удаление существующего представления:
+```bash
+echo "DROP TABLE IF EXISTS mining.mv_hash_rate_stats" | curl 'http://localhost:8123/?database=mining' --data-binary @- -u default:5555
+```
+
+Создание нового представления из файла:
+```bash
+curl 'http://localhost:8123/?user=default&password=5555&database=mining' --data-binary @/home/ro/projects/nomium/dev/stratum/nomium/shares-logger/src/storage/clickhouse/queries/hashrate_view.sql
+```
+
+Проверка что представление создалось:
+```bash
+echo "SELECT name, engine FROM system.tables WHERE database = 'mining' AND name = 'mv_hash_rate_stats'" | curl 'http://localhost:8123/?database=mining' --data-binary @- -u default:5555
+```
+
+3. Базовые запросы к MV:
+
+// в связи работами над уточнением хэшрейта запросы и структура MV сильно изменятся, ожидаем.  

nomium/shares-logger/src/lib.rs

@@ -4,9 +4,9 @@ pub mod models;
 pub mod services;
 pub mod storage;
 pub mod traits;
+pub mod worker_name_store;
 
 use crate::config::SETTINGS;
-use crate::traits::ShareData;
 use log::info;
 use std::sync::Arc;
 use tokio::sync::{mpsc::{self, error::TrySendError}, Mutex};
@@ -18,6 +18,8 @@ use crate::traits::ShareStorage;
 use crate::storage::clickhouse::ClickhouseBlockStorage;
 use crate::models::BlockFound;
 use std::time::Instant;
+use serde::Serialize;
+use serde::de::DeserializeOwned;
 
 lazy_static! {
     static ref GLOBAL_LOGGER: ShareLogger<ShareLog> = {
@@ -45,18 +47,18 @@ pub fn log_block(block: BlockFound) {
     BLOCK_LOGGER.log_share(block);
 }
 
-pub struct ShareLogger<T: ShareData> {
+pub struct ShareLogger<T: Send + Sync + Clone + Serialize + DeserializeOwned> {
     primary_tx: mpsc::Sender<T>,
     backup_tx: mpsc::UnboundedSender<T>,
 }
 
-pub struct ShareLoggerBuilder<T: ShareData> {
+pub struct ShareLoggerBuilder<T: Send + Sync + Clone + Serialize + DeserializeOwned> {
     storage: Arc<Mutex<Box<dyn ShareStorage<T>>>>,
     primary_channel_size: Option<usize>,
     backup_check_interval: Option<Duration>,
 }
 
-impl<T: ShareData + 'static> ShareLoggerBuilder<T> {
+impl<T: Send + Sync + Clone + Serialize + DeserializeOwned + 'static> ShareLoggerBuilder<T> {
 
     pub fn new(storage: Box<dyn ShareStorage<T>>) -> Self {
         Self {
@@ -103,7 +105,7 @@ impl<T: ShareData + 'static> ShareLoggerBuilder<T> {
     }
 }
 
-impl<T: ShareData + 'static> ShareLogger<T> {
+impl<T: Send + Sync + Clone + Serialize + DeserializeOwned + 'static> ShareLogger<T> {
     pub fn log_share(&self, share: T) {
         match self.primary_tx.try_send(share.clone()) {
             Ok(_) => (),
@@ -116,7 +118,7 @@ impl<T: ShareData + 'static> ShareLogger<T> {
     }
 }
 
-async fn process_shares<T: ShareData>(
+async fn process_shares<T: Send + Sync + Clone + Serialize + DeserializeOwned>(
     mut primary_rx: mpsc::Receiver<T>,
     mut backup_rx: mpsc::UnboundedReceiver<T>,
     storage: Arc<Mutex<Box<dyn ShareStorage<T>>>>,
@@ -129,36 +131,20 @@ async fn process_shares<T: ShareData>(
     }
     let init_duration = init_start.elapsed();
     info!("Storage initialized in: {:?}", init_duration);
-
     let mut backup_interval = tokio::time::interval(backup_check_interval);
-
     loop {
         tokio::select! {
             Some(share) = primary_rx.recv() => {
                 info!("Processing share from primary channel");
-
-                if share.is_block_found() {
-                    if let Err(e) = storage.lock().await.store_share(share).await {
-                        info!("Failed to store block: {}", e);
-                    }
-                } else {
-                    if let Err(e) = storage.lock().await.store_share(share).await {
-                        info!("Failed to store share: {}", e);
-                    }
+                if let Err(e) = storage.lock().await.store_share(share).await {
+                    info!("Failed to store share: {}", e);
                 }
             }
             _ = backup_interval.tick() => {
                 let mut backup_shares = Vec::new();
                 while let Ok(share) = backup_rx.try_recv() {
-                    if share.is_block_found() {
-                        if let Err(e) = storage.lock().await.store_share(share).await {
-                            info!("Failed to store backup block: {}", e);
-                        }
-                    } else {
-                        backup_shares.push(share);
-                    }
+                    backup_shares.push(share);
                 }
-
                 if !backup_shares.is_empty() {
                     if let Err(e) = storage.lock().await.store_batch(backup_shares).await {
                         info!("Failed to store backup shares: {}", e);

nomium/shares-logger/src/models/block_found.rs

@@ -1,31 +1,50 @@
 use serde::{Serialize, Deserialize};
-use crate::traits::ShareData;
-use async_trait::async_trait;
+use serde_json::Value;
+use serde_json::json;
+use log::info;
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct BlockFound {
     pub channel_id: u32,
     pub block_hash: Vec<u8>,
     pub ntime: u32,
+    pub user_identity: String,
+    pub worker_id: String,
 }
 
-#[async_trait]
-impl ShareData for BlockFound {
-    fn get_identifier(&self) -> String {
-        format!("{}_{}_{}", self.channel_id, hex::encode(&self.block_hash), self.ntime)
-    }
+impl BlockFound {
+    pub fn prepare_block(
+        channel_id: u32,
+        block_hash: Vec<u8>,
+        ntime: u32,
+        user_identity_json: String,
+    ) -> Self {
+        info!("Preparing block with user_identity_json: {}", user_identity_json);
+        
+        let worker_identity: Value = serde_json::from_str(&user_identity_json)
+            .unwrap_or_else(|_| json!({
+                "worker_name": user_identity_json.clone(),
+                "worker_id": "unknown"
+            }));
 
-    async fn validate(&self) -> bool {
-        self.block_hash.len() == 32
-    }
+        let user_identity = worker_identity["worker_name"]
+            .as_str()
+            .unwrap_or(&user_identity_json)
+            .to_string();
 
-    fn to_storage_format(&self) -> Vec<(String, String)> {
-        vec![
-            ("channel_id".to_string(), self.channel_id.to_string()),
-            ("block_hash".to_string(), hex::encode(&self.block_hash)),
-            ("ntime".to_string(), self.ntime.to_string()),
-        ]
-    }
+        let worker_id = worker_identity["worker_id"]
+            .as_str()
+            .unwrap_or("unknown")
+            .to_string();
 
-    fn is_block_found(&self) -> bool { true }
+        info!("Block prepared with worker_id: {}", worker_id);
+
+        BlockFound {
+            channel_id,
+            block_hash,
+            ntime,
+            user_identity,
+            worker_id,
+        }
+    }
 }

nomium/shares-logger/src/models/clickhouse_block.rs

@@ -2,11 +2,13 @@ use crate::models::BlockFound;
 use clickhouse::Row;
 use serde::Serialize;
 
-#[derive(Row, Serialize)]
+#[derive(Row, Serialize, Debug)]
 pub struct ClickhouseBlock {
     pub channel_id: u32,
     pub block_hash: String,
     pub ntime: u32,
+    pub user_identity: String,
+    pub worker_id: String,
 }
 
 impl From<BlockFound> for ClickhouseBlock {
@@ -15,6 +17,8 @@ impl From<BlockFound> for ClickhouseBlock {
             channel_id: block.channel_id,
             block_hash: hex::encode(&block.block_hash),
             ntime: block.ntime,
+            user_identity: block.user_identity,
+            worker_id: block.worker_id,
         }
     }
 }

nomium/shares-logger/src/models/clickhouse_share.rs

@@ -14,6 +14,8 @@ pub struct ClickhouseShare {
     pub share_status: u8,
     pub extranonce: String,
     pub difficulty: f64,
+    pub user_identity: String,
+    pub worker_id: String,
 }
 
 impl From<ShareLog> for ClickhouseShare {
@@ -42,6 +44,8 @@ impl From<ShareLog> for ClickhouseShare {
             share_status: share.share_status as u8,
             extranonce: extranonce_hex,
             difficulty: share.difficulty,
+            user_identity: share.user_identity,
+            worker_id: share.worker_id,
         }
     }
 }

nomium/shares-logger/src/models/share_log.rs

@@ -1,6 +1,4 @@
 use serde::{Serialize, Deserialize};
-use crate::traits::ShareData;
-use async_trait::async_trait;
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct ShareLog {
@@ -14,6 +12,8 @@ pub struct ShareLog {
     pub share_status: ShareStatus,
     pub extranonce: Vec<u8>,
     pub difficulty: f64,
+    pub user_identity: String,
+    pub worker_id: String,
 }
 
 #[derive(Debug, Clone, Copy, Serialize, Deserialize)]
@@ -36,6 +36,8 @@ impl ShareLog {
         share_status: ShareStatus,
         extranonce: Vec<u8>,
         difficulty: f64,
+        user_identity: String,
+        worker_id: String,
     ) -> Self {
         Self {
             channel_id,
@@ -47,35 +49,9 @@ impl ShareLog {
             hash,
             share_status,
             extranonce,
-            difficulty
+            difficulty,
+            user_identity,
+            worker_id,
         }
     }
-}
-
-#[async_trait]
-impl ShareData for ShareLog {
-    fn get_identifier(&self) -> String {
-        format!("{}_{}", self.channel_id, self.sequence_number)
-    }
-
-    async fn validate(&self) -> bool {
-        true
-    }
-
-    fn to_storage_format(&self) -> Vec<(String, String)> {
-        vec![
-            ("channel_id".to_string(), self.channel_id.to_string()),
-            ("sequence_number".to_string(), self.sequence_number.to_string()),
-            ("job_id".to_string(), self.job_id.to_string()),
-            ("nonce".to_string(), self.nonce.to_string()),
-            ("ntime".to_string(), self.ntime.to_string()),
-            ("version".to_string(), self.version.to_string()),
-            ("hash".to_string(), hex::encode(&self.hash)),
-            ("share_status".to_string(), (self.share_status as u8).to_string()),
-            ("extranonce".to_string(), hex::encode(&self.extranonce)),
-            ("difficulty".to_string(), self.difficulty.to_string()),
-        ]
-    }
-
-    fn is_block_found(&self) -> bool { false }
 }

nomium/shares-logger/src/services/share_processor.rs

@@ -2,6 +2,9 @@ use crate::models::ShareLog;
 use mining_sv2::Target;
 use super::difficulty::DifficultyService;
 use crate::models::ShareStatus;
+use log::info;
+use serde_json::Value;
+use serde_json::json;
 
 pub struct ShareProcessor;
 
@@ -15,8 +18,26 @@ impl ShareProcessor {
         version: u32,
         hash: [u8; 32],
         downstream_target: Target,
-        extranonce: Vec<u8>,  // Принимаем Vec<u8>
+        extranonce: Vec<u8>,
+        user_identity_json: String,
     ) -> ShareLog {
+        let worker_identity: Value = serde_json::from_str(&user_identity_json)
+            .unwrap_or_else(|_| json!({
+                "worker_name": user_identity_json.clone(),
+                "worker_id": "unknown"
+            }));
+
+        let user_identity = worker_identity["worker_name"]
+            .as_str()
+            .unwrap_or(&user_identity_json)
+            .to_string();
+        
+        let worker_id = worker_identity["worker_id"]
+            .as_str()
+            .unwrap_or("unknown")
+            .to_string();
+
+        info!("user_identity from prepare_share_log: {}", user_identity);
         let mut hash_bytes = hash;
         hash_bytes.reverse();
         let difficulty = DifficultyService::calculate_difficulty_from_hash(&hash_bytes);
@@ -34,6 +55,8 @@ impl ShareProcessor {
             status,
             extranonce,
             difficulty,
+            user_identity,
+            worker_id,
         )
     }
 }